mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
refactor: [BREAKING] Remove generic Workflow<T> (#1551)
Remove input type checking in favour of explicit `.DescribeProtocolAsync()` flow. Also removes `.AsAgentAsync()` as the validation happens at workflow run time. This makes it easier to use Workflows with DI without resorting to async-over-sync.
This commit is contained in:
Jacob Alber
committed by
GitHub
Unverified
GitHub
parent
7c1e3db846
commit
0ef4e739d5
@@ -36,6 +36,18 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen
|
||||
return runner.ResumeStreamAsync(this.ExecutionMode, fromCheckpoint, cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask<StreamingRun> StreamAsync(
|
||||
Workflow workflow,
|
||||
string? runId = null,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [], cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
return new(runHandle);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask<StreamingRun> StreamAsync<TInput>(
|
||||
Workflow workflow,
|
||||
@@ -50,16 +62,17 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask<StreamingRun> StreamAsync<TInput>(
|
||||
Workflow<TInput> workflow,
|
||||
TInput input,
|
||||
public async ValueTask<Checkpointed<StreamingRun>> StreamAsync(
|
||||
Workflow workflow,
|
||||
CheckpointManager checkpointManager,
|
||||
string? runId = null,
|
||||
CancellationToken cancellationToken = default) where TInput : notnull
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [typeof(TInput)], cancellationToken)
|
||||
AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager, runId: runId, [], cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
return await runHandle.EnqueueAndStreamAsync(input, cancellationToken).ConfigureAwait(false);
|
||||
return await runHandle.WithCheckpointingAsync<StreamingRun>(() => new(new StreamingRun(runHandle)))
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
@@ -77,21 +90,6 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask<Checkpointed<StreamingRun>> StreamAsync<TInput>(
|
||||
Workflow<TInput> workflow,
|
||||
TInput input,
|
||||
CheckpointManager checkpointManager,
|
||||
string? runId = null,
|
||||
CancellationToken cancellationToken = default) where TInput : notnull
|
||||
{
|
||||
AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager, runId: runId, [typeof(TInput)], cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
return await runHandle.WithCheckpointingAsync(() => runHandle.EnqueueAndStreamAsync(input, cancellationToken))
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask<Checkpointed<StreamingRun>> ResumeStreamAsync(
|
||||
Workflow workflow,
|
||||
@@ -107,19 +105,24 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask<Checkpointed<StreamingRun>> ResumeStreamAsync<TInput>(
|
||||
Workflow<TInput> workflow,
|
||||
CheckpointInfo fromCheckpoint,
|
||||
CheckpointManager checkpointManager,
|
||||
private async ValueTask<AsyncRunHandle> BeginRunHandlingChatProtocolAsync<TInput>(Workflow workflow,
|
||||
TInput input,
|
||||
CheckpointManager? checkpointManager,
|
||||
string? runId = null,
|
||||
CancellationToken cancellationToken = default) where TInput : notnull
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
AsyncRunHandle runHandle = await this.ResumeRunAsync(workflow, checkpointManager, runId: runId, fromCheckpoint, [typeof(TInput)], cancellationToken)
|
||||
ProtocolDescriptor descriptor = await workflow.DescribeProtocolAsync(cancellationToken).ConfigureAwait(false);
|
||||
AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager, runId, descriptor.Accepts, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
return await runHandle.WithCheckpointingAsync<StreamingRun>(() => new(new StreamingRun(runHandle)))
|
||||
.ConfigureAwait(false);
|
||||
await runHandle.EnqueueMessageAsync(input, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if (descriptor.IsChatProtocol() && input is not TurnToken)
|
||||
{
|
||||
await runHandle.EnqueueMessageAsync(new TurnToken(emitEvents: true), cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return runHandle;
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
@@ -129,21 +132,13 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen
|
||||
string? runId = null,
|
||||
CancellationToken cancellationToken = default) where TInput : notnull
|
||||
{
|
||||
var runHandle = await this.GetRunHandleWithTurnTokenAsync(workflow: workflow, input: input, checkpointManager: null, runId: runId, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
Run run = new(runHandle);
|
||||
await run.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false);
|
||||
return run;
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask<Run> RunAsync<TInput>(
|
||||
Workflow<TInput> workflow,
|
||||
TInput input,
|
||||
string? runId = null,
|
||||
CancellationToken cancellationToken = default) where TInput : notnull
|
||||
{
|
||||
var runHandle = await this.GetRunHandleWithTurnTokenAsync(workflow: workflow, input: input, checkpointManager: null, runId: runId, cancellationToken).ConfigureAwait(false);
|
||||
AsyncRunHandle runHandle = await this.BeginRunHandlingChatProtocolAsync(
|
||||
workflow,
|
||||
input,
|
||||
checkpointManager: null,
|
||||
runId,
|
||||
cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
Run run = new(runHandle);
|
||||
await run.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false);
|
||||
@@ -158,23 +153,13 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen
|
||||
string? runId = null,
|
||||
CancellationToken cancellationToken = default) where TInput : notnull
|
||||
{
|
||||
var runHandle = await this.GetRunHandleWithTurnTokenAsync(workflow: workflow, input: input, checkpointManager: checkpointManager, runId: runId, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
Run run = new(runHandle);
|
||||
await run.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false);
|
||||
return await runHandle.WithCheckpointingAsync(() => new ValueTask<Run>(run))
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask<Checkpointed<Run>> RunAsync<TInput>(
|
||||
Workflow<TInput> workflow,
|
||||
TInput input,
|
||||
CheckpointManager checkpointManager,
|
||||
string? runId = null,
|
||||
CancellationToken cancellationToken = default) where TInput : notnull
|
||||
{
|
||||
var runHandle = await this.GetRunHandleWithTurnTokenAsync(workflow: workflow, input: input, checkpointManager: checkpointManager, runId: runId, cancellationToken).ConfigureAwait(false);
|
||||
AsyncRunHandle runHandle = await this.BeginRunHandlingChatProtocolAsync(
|
||||
workflow,
|
||||
input,
|
||||
checkpointManager,
|
||||
runId,
|
||||
cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
Run run = new(runHandle);
|
||||
await run.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false);
|
||||
@@ -196,63 +181,4 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen
|
||||
return await runHandle.WithCheckpointingAsync<Run>(() => new(new Run(runHandle)))
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask<Checkpointed<Run>> ResumeAsync<TInput>(
|
||||
Workflow<TInput> workflow,
|
||||
CheckpointInfo fromCheckpoint,
|
||||
CheckpointManager checkpointManager,
|
||||
string? runId = null,
|
||||
CancellationToken cancellationToken = default) where TInput : notnull
|
||||
{
|
||||
AsyncRunHandle runHandle = await this.ResumeRunAsync(workflow, checkpointManager, runId: runId, fromCheckpoint, [typeof(TInput)], cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
return await runHandle.WithCheckpointingAsync<Run>(() => new(new Run(runHandle)))
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// Helper to construct a RunHandle with the provided input enqueued. If the starting executor supports it, a TurnToken will be enqueued also.
|
||||
private async ValueTask<AsyncRunHandle> GetRunHandleWithTurnTokenAsync<TInput>(
|
||||
Workflow workflow,
|
||||
TInput input,
|
||||
CheckpointManager? checkpointManager,
|
||||
string? runId,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var knownTypes = new List<Type>() { typeof(TInput) };
|
||||
var needsTurnToken = await StartingExecutorHandlesTurnTokenAsync<TInput>(workflow).ConfigureAwait(false);
|
||||
if (needsTurnToken)
|
||||
{
|
||||
knownTypes.Add(typeof(TurnToken));
|
||||
}
|
||||
|
||||
AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: checkpointManager, runId: runId, knownTypes, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
await runHandle.EnqueueMessageAsync(input, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if (needsTurnToken)
|
||||
{
|
||||
await runHandle.EnqueueMessageAsync(new TurnToken(emitEvents: true), cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return runHandle;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Helper method to detect if the starting executor of a given workflow accepts the provided input type as well as a TurnToken.
|
||||
/// </summary>
|
||||
private static async ValueTask<bool> StartingExecutorHandlesTurnTokenAsync<TInput>(Workflow workflow)
|
||||
{
|
||||
if (workflow.Registrations.TryGetValue(workflow.StartExecutorId, out var registration))
|
||||
{
|
||||
// Create instance to check type
|
||||
Executor startExecutor = await registration.CreateInstanceAsync(string.Empty)
|
||||
.ConfigureAwait(false);
|
||||
return startExecutor.CanHandle(typeof(TInput)) && startExecutor.CanHandle(typeof(TurnToken));
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user