From a3cca78592dd24c5658397d6807fc5528158fb11 Mon Sep 17 00:00:00 2001 From: luoqiang <2769838458@qq.com> Date: Fri, 28 Aug 2026 10:43:57 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8B=86=E5=88=86=20CLI=20=E4=B8=8E?= =?UTF-8?q?=E7=BD=91=E9=A1=B5=E5=AE=BF=E4=B8=BB=EF=BC=8C=E5=B9=B6=E5=9C=A8?= =?UTF-8?q?=E6=97=A0=E6=9C=89=E6=95=88=E5=9F=8E=E5=B8=82=E6=97=B6=E6=9A=82?= =?UTF-8?q?=E5=81=9C=E7=AD=89=E4=BA=BA=E7=A1=AE=E8=AE=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将原单体 MAF1 拆成 MAF1（控制台工作流）和 MAF1.Web（可视化编排）。 抽城市失败时暂停：网页弹确认、CLI 控制台询问，超时采用默认结束查询； 也可改填城市后继续。共享决策模型放在 MAF1.Core。 --- MAF1.Core/Agents/AgentStepResult.cs | 2 +- MAF1.Core/Decisions/DecisionModels.cs | 97 +++ {Data => MAF1.Web/Data}/cities.txt | 0 {Data => MAF1.Web/Data}/not-cities.txt | 0 MAF1.csproj => MAF1.Web/MAF1.Web.csproj | 34 +- .../PluginHost}/CredentialStore.cs | 0 .../PluginHost}/NodeCatalogService.cs | 0 .../PluginHost}/PluginOptions.cs | 0 .../PluginHost}/PluginProcessRunner.cs | 0 .../PluginHost}/PluginScanner.cs | 0 Program.cs => MAF1.Web/Program.cs | 17 +- MAF1.Web/Properties/launchSettings.json | 14 + {Web => MAF1.Web/Web}/AgentCatalog.cs | 0 {Web => MAF1.Web/Web}/AgentRuntime.cs | 6 +- MAF1.Web/Web/ConfigurableWorkflowRunner.cs | 582 ++++++++++++++++++ {Web => MAF1.Web/Web}/WorkflowModels.cs | 13 + MAF1.Web/Web/WorkflowRunStore.cs | 35 ++ appsettings.json => MAF1.Web/appsettings.json | 3 + {wwwroot => MAF1.Web/wwwroot}/css/app.css | 90 +++ {wwwroot => MAF1.Web/wwwroot}/index.html | 15 + {wwwroot => MAF1.Web/wwwroot}/js/app.js | 170 ++++- MAF1.slnx | 3 +- CliHost.cs => MAF1/CliHost.cs | 22 +- MAF1/Data/cities.txt | 3 + MAF1/Data/not-cities.txt | 3 + MAF1/MAF1.csproj | 32 + MAF1/Orchestration/ConsoleDecisionPrompt.cs | 91 +++ .../Orchestration}/WorkflowOrchestration.cs | 17 + MAF1/Program.cs | 5 + .../Properties}/launchSettings.json | 12 +- MAF1/Workflows/ApplyCityDecisionExecutor.cs | 42 ++ MAF1/Workflows/AskCityDecisionExecutor.cs | 26 + .../Workflows}/CityGateExecutor.cs | 13 +- .../Workflows}/CityParseExecutor.cs | 0 .../Workflows}/CityWeatherWorkflow.cs | 27 +- .../Workflows}/ToWeatherPromptExecutor.cs | 0 app.manifest => MAF1/app.manifest | 0 MAF1/appsettings.json | 21 + Plugins/README.md | 2 +- README.md | 172 ++++-- Web/ConfigurableWorkflowRunner.cs | 287 --------- Workflows/SkipWeatherExecutor.cs | 28 - 42 files changed, 1427 insertions(+), 457 deletions(-) create mode 100644 MAF1.Core/Decisions/DecisionModels.cs rename {Data => MAF1.Web/Data}/cities.txt (100%) rename {Data => MAF1.Web/Data}/not-cities.txt (100%) rename MAF1.csproj => MAF1.Web/MAF1.Web.csproj (52%) rename {PluginHost => MAF1.Web/PluginHost}/CredentialStore.cs (100%) rename {PluginHost => MAF1.Web/PluginHost}/NodeCatalogService.cs (100%) rename {PluginHost => MAF1.Web/PluginHost}/PluginOptions.cs (100%) rename {PluginHost => MAF1.Web/PluginHost}/PluginProcessRunner.cs (100%) rename {PluginHost => MAF1.Web/PluginHost}/PluginScanner.cs (100%) rename Program.cs => MAF1.Web/Program.cs (69%) create mode 100644 MAF1.Web/Properties/launchSettings.json rename {Web => MAF1.Web/Web}/AgentCatalog.cs (100%) rename {Web => MAF1.Web/Web}/AgentRuntime.cs (83%) create mode 100644 MAF1.Web/Web/ConfigurableWorkflowRunner.cs rename {Web => MAF1.Web/Web}/WorkflowModels.cs (75%) create mode 100644 MAF1.Web/Web/WorkflowRunStore.cs rename appsettings.json => MAF1.Web/appsettings.json (92%) rename {wwwroot => MAF1.Web/wwwroot}/css/app.css (71%) rename {wwwroot => MAF1.Web/wwwroot}/index.html (64%) rename {wwwroot => MAF1.Web/wwwroot}/js/app.js (69%) rename CliHost.cs => MAF1/CliHost.cs (79%) create mode 100644 MAF1/Data/cities.txt create mode 100644 MAF1/Data/not-cities.txt create mode 100644 MAF1/MAF1.csproj create mode 100644 MAF1/Orchestration/ConsoleDecisionPrompt.cs rename {Orchestration => MAF1/Orchestration}/WorkflowOrchestration.cs (70%) create mode 100644 MAF1/Program.cs rename {Properties => MAF1/Properties}/launchSettings.json (58%) create mode 100644 MAF1/Workflows/ApplyCityDecisionExecutor.cs create mode 100644 MAF1/Workflows/AskCityDecisionExecutor.cs rename {Workflows => MAF1/Workflows}/CityGateExecutor.cs (77%) rename {Workflows => MAF1/Workflows}/CityParseExecutor.cs (100%) rename {Workflows => MAF1/Workflows}/CityWeatherWorkflow.cs (56%) rename {Workflows => MAF1/Workflows}/ToWeatherPromptExecutor.cs (100%) rename app.manifest => MAF1/app.manifest (100%) create mode 100644 MAF1/appsettings.json delete mode 100644 Web/ConfigurableWorkflowRunner.cs delete mode 100644 Workflows/SkipWeatherExecutor.cs diff --git a/MAF1.Core/Agents/AgentStepResult.cs b/MAF1.Core/Agents/AgentStepResult.cs index 6db28cc..72a806a 100644 --- a/MAF1.Core/Agents/AgentStepResult.cs +++ b/MAF1.Core/Agents/AgentStepResult.cs @@ -9,7 +9,7 @@ public sealed class AgentStepResult public Dictionary Outputs { get; init; } = new(StringComparer.OrdinalIgnoreCase); } -internal static class AgentInputs +public static class AgentInputs { public static string? ReadString(IReadOnlyDictionary inputs, string key) { diff --git a/MAF1.Core/Decisions/DecisionModels.cs b/MAF1.Core/Decisions/DecisionModels.cs new file mode 100644 index 0000000..8bba520 --- /dev/null +++ b/MAF1.Core/Decisions/DecisionModels.cs @@ -0,0 +1,97 @@ +using MAF1.Agents; + +namespace MAF1.Decisions; + +public sealed class DecisionRequest +{ + public string Id { get; set; } = ""; + public string NodeId { get; set; } = ""; + public string Prompt { get; set; } = ""; + public string? Reason { get; set; } + public List Options { get; set; } = []; + public string DefaultOptionId { get; set; } = ""; + public int TimeoutSeconds { get; set; } + public DateTimeOffset Deadline { get; set; } +} + +public sealed class DecisionOption +{ + public string Id { get; set; } = ""; + public string Label { get; set; } = ""; + public string? Description { get; set; } + public bool RequiresText { get; set; } + public string? TextPlaceholder { get; set; } + public bool IsDefault { get; set; } +} + +public sealed class DecisionAnswer +{ + public string OptionId { get; set; } = ""; + public string? Text { get; set; } + public bool TimedOut { get; set; } +} + +public static class DecisionOptionIds +{ + public const string Stop = "stop"; + public const string QueryCities = "query-cities"; +} + +public static class EmptyCityDecision +{ + public static DecisionRequest Create(string nodeId, string? reason, int timeoutSeconds) + { + int seconds = Math.Clamp(timeoutSeconds, 1, 600); + return new DecisionRequest + { + Id = Guid.NewGuid().ToString("n")[..12], + NodeId = nodeId, + Reason = reason, + Prompt = "没有读到有效城市。默认结束后续查询；也可以改填其它城市后继续。超时将自动采用默认方案。", + DefaultOptionId = DecisionOptionIds.Stop, + TimeoutSeconds = seconds, + Deadline = DateTimeOffset.UtcNow.AddSeconds(seconds), + Options = + [ + new DecisionOption + { + Id = DecisionOptionIds.Stop, + Label = "结束查询（默认）", + Description = "保持当前结果，跳过需要城市的后续节点。", + IsDefault = true, + }, + new DecisionOption + { + Id = DecisionOptionIds.QueryCities, + Label = "改查其它城市", + Description = "手动输入城市名，下游天气节点会按这份名单继续。", + RequiresText = true, + TextPlaceholder = "例如：成都、北京", + }, + ], + }; + } + + public static bool TryContinueWithCities(DecisionAnswer answer, out List cities) + { + cities = []; + if (answer.TimedOut) + { + return false; + } + + bool queryCities = answer.OptionId.Equals(DecisionOptionIds.QueryCities, StringComparison.OrdinalIgnoreCase); + if (!queryCities) + { + return false; + } + + cities = ParseCities(answer.Text); + return cities.Count > 0; + } + + public static List ParseCities(string? text) + => AgentInputs.ReadStringList( + new Dictionary(StringComparer.OrdinalIgnoreCase) { ["cities"] = text }, + "cities"); +} diff --git a/Data/cities.txt b/MAF1.Web/Data/cities.txt similarity index 100% rename from Data/cities.txt rename to MAF1.Web/Data/cities.txt diff --git a/Data/not-cities.txt b/MAF1.Web/Data/not-cities.txt similarity index 100% rename from Data/not-cities.txt rename to MAF1.Web/Data/not-cities.txt diff --git a/MAF1.csproj b/MAF1.Web/MAF1.Web.csproj similarity index 52% rename from MAF1.csproj rename to MAF1.Web/MAF1.Web.csproj index cd31d99..2afc4f6 100644 --- a/MAF1.csproj +++ b/MAF1.Web/MAF1.Web.csproj @@ -1,38 +1,18 @@ - + - Exe net10.0 enable enable - app.manifest + MAF1 - - - - - - - - - - - - - - - - - - - - - + + false - + false @@ -48,8 +28,8 @@ - <_FileCityOut>$(MSBuildThisFileDirectory)plugins\file-city\bin\$(Configuration)\net10.0 - <_WeatherOut>$(MSBuildThisFileDirectory)plugins\weather\bin\$(Configuration)\net10.0 + <_FileCityOut>$(MSBuildThisFileDirectory)..\plugins\file-city\bin\$(Configuration)\net10.0 + <_WeatherOut>$(MSBuildThisFileDirectory)..\plugins\weather\bin\$(Configuration)\net10.0 <_FileCityFiles Include="$(_FileCityOut)\**\*" /> diff --git a/PluginHost/CredentialStore.cs b/MAF1.Web/PluginHost/CredentialStore.cs similarity index 100% rename from PluginHost/CredentialStore.cs rename to MAF1.Web/PluginHost/CredentialStore.cs diff --git a/PluginHost/NodeCatalogService.cs b/MAF1.Web/PluginHost/NodeCatalogService.cs similarity index 100% rename from PluginHost/NodeCatalogService.cs rename to MAF1.Web/PluginHost/NodeCatalogService.cs diff --git a/PluginHost/PluginOptions.cs b/MAF1.Web/PluginHost/PluginOptions.cs similarity index 100% rename from PluginHost/PluginOptions.cs rename to MAF1.Web/PluginHost/PluginOptions.cs diff --git a/PluginHost/PluginProcessRunner.cs b/MAF1.Web/PluginHost/PluginProcessRunner.cs similarity index 100% rename from PluginHost/PluginProcessRunner.cs rename to MAF1.Web/PluginHost/PluginProcessRunner.cs diff --git a/PluginHost/PluginScanner.cs b/MAF1.Web/PluginHost/PluginScanner.cs similarity index 100% rename from PluginHost/PluginScanner.cs rename to MAF1.Web/PluginHost/PluginScanner.cs diff --git a/Program.cs b/MAF1.Web/Program.cs similarity index 69% rename from Program.cs rename to MAF1.Web/Program.cs index c3366b6..fcc4125 100644 --- a/Program.cs +++ b/MAF1.Web/Program.cs @@ -1,15 +1,9 @@ -using MAF1; +using MAF1.Decisions; using MAF1.Utils; using MAF1.Web; WindowsConsole.EnableUtf8(); -if (args.Length > 0 && args[0] is "node" or "edge" or "--cli" or "-h" or "--help") -{ - await CliHost.RunAsync(args); - return; -} - WebApplicationBuilder builder = WebApplication.CreateBuilder(args); builder.WebHost.UseUrls("http://127.0.0.1:5288"); builder.Services.ConfigureHttpJsonOptions(options => @@ -42,6 +36,15 @@ app.MapGet("/api/status", (AgentRuntime runtime) => new }); app.MapPost("/api/run", (WorkflowGraph graph, AgentRuntime runtime, CancellationToken cancellationToken) => runtime.Runner.RunAsync(graph, cancellationToken)); +app.MapGet("/api/run/{runId}", (string runId, AgentRuntime runtime) => +{ + WorkflowRunResult? result = runtime.Runner.Get(runId); + return result is null + ? Results.NotFound(new WorkflowRunResult { Ok = false, Status = WorkflowRunStatus.Failed, Error = "找不到这次运行。" }) + : Results.Ok(result); +}); +app.MapPost("/api/run/{runId}/decide", (string runId, DecisionAnswer answer, AgentRuntime runtime, CancellationToken cancellationToken) + => runtime.Runner.DecideAsync(runId, answer, cancellationToken)); Console.WriteLine("可视化编排: http://127.0.0.1:5288"); app.Run(); diff --git a/MAF1.Web/Properties/launchSettings.json b/MAF1.Web/Properties/launchSettings.json new file mode 100644 index 0000000..740adaf --- /dev/null +++ b/MAF1.Web/Properties/launchSettings.json @@ -0,0 +1,14 @@ +{ + "profiles": { + "designer": { + "commandName": "Project", + "applicationUrl": "http://127.0.0.1:5288", + "environmentVariables": { + "ASPNETCORE_URLS": "http://127.0.0.1:5288", + "OPENAI_ENDPOINT": "https://api.deepseek.com", + "OPENAI_API_KEY": "sk-b31330cf3a414d86ab614b9dfb3650c3", + "OPENAI_CHAT_MODEL": "deepseek-v4-flash" + } + } + } +} diff --git a/Web/AgentCatalog.cs b/MAF1.Web/Web/AgentCatalog.cs similarity index 100% rename from Web/AgentCatalog.cs rename to MAF1.Web/Web/AgentCatalog.cs diff --git a/Web/AgentRuntime.cs b/MAF1.Web/Web/AgentRuntime.cs similarity index 83% rename from Web/AgentRuntime.cs rename to MAF1.Web/Web/AgentRuntime.cs index 97d7ecf..2ab3d08 100644 --- a/Web/AgentRuntime.cs +++ b/MAF1.Web/Web/AgentRuntime.cs @@ -23,7 +23,9 @@ public sealed class AgentRuntime Scanner = new PluginScanner(pluginOptions); Catalog = new NodeCatalogService(Scanner); PluginRunner = new PluginProcessRunner(pluginOptions, Credentials); - Runner = new ConfigurableWorkflowRunner(FileCity, Weather, Catalog, PluginRunner); + RunStore = new WorkflowRunStore(); + DecisionTimeoutSeconds = Math.Clamp(config.GetValue("Workflow:DecisionTimeoutSeconds", 30), 1, 600); + Runner = new ConfigurableWorkflowRunner(FileCity, Weather, Catalog, PluginRunner, RunStore, DecisionTimeoutSeconds); WeatherProvider = weatherOptions.Provider; PluginsRoot = pluginOptions.ResolveRoot(); } @@ -31,6 +33,8 @@ public sealed class AgentRuntime public AIAgent FileCity { get; } public AIAgent Weather { get; } public ConfigurableWorkflowRunner Runner { get; } + public WorkflowRunStore RunStore { get; } + public int DecisionTimeoutSeconds { get; } public NodeCatalogService Catalog { get; } public PluginScanner Scanner { get; } public PluginProcessRunner PluginRunner { get; } diff --git a/MAF1.Web/Web/ConfigurableWorkflowRunner.cs b/MAF1.Web/Web/ConfigurableWorkflowRunner.cs new file mode 100644 index 0000000..2721ba3 --- /dev/null +++ b/MAF1.Web/Web/ConfigurableWorkflowRunner.cs @@ -0,0 +1,582 @@ +using System.Text.Json; +using MAF1.Agents; +using MAF1.Agents.FileCity; +using MAF1.Agents.Weather; +using MAF1.Decisions; +using MAF1.PluginContract; +using MAF1.Plugins; +using Microsoft.Agents.AI; + +namespace MAF1.Web; + +public sealed class ConfigurableWorkflowRunner( + AIAgent fileCityAgent, + AIAgent weatherAgent, + NodeCatalogService catalog, + PluginProcessRunner plugins, + WorkflowRunStore runs, + int decisionTimeoutSeconds) +{ + private static readonly HashSet ReservedConfigKeys = new(StringComparer.OrdinalIgnoreCase) + { + "credentialId", + "askOnEmptyCities", + "decisionTimeoutSeconds", + }; + + public async Task RunAsync(WorkflowGraph graph, CancellationToken cancellationToken) + { + try + { + Validate(graph); + NodeCatalogSnapshot snapshot = catalog.Load(); + List order = TopologicalOrder(graph); + WorkflowSession session = new() + { + RunId = Guid.NewGuid().ToString("n")[..12], + Graph = graph, + Catalog = snapshot, + Order = order, + Outputs = new Dictionary>(StringComparer.OrdinalIgnoreCase), + Steps = [], + NextIndex = 0, + }; + runs.TryAdd(session); + return await ContinueAsync(session, cancellationToken); + } + catch (Exception ex) + { + return Fail(ex.Message); + } + } + + public WorkflowRunResult? Get(string runId) => runs.GetResult(runId); + + public async Task DecideAsync(string runId, DecisionAnswer answer, CancellationToken cancellationToken) + { + if (!runs.TryGet(runId, out WorkflowSession? session) || session is null) + { + return Fail("找不到这次运行，可能已经结束或服务已重启。"); + } + + return await ResolveAsync(session, answer, cancellationToken, session.Pending?.Id); + } + + private async Task ContinueAsync(WorkflowSession session, CancellationToken cancellationToken) + { + try + { + for (int i = session.NextIndex; i < session.Order.Count; i++) + { + string nodeId = session.Order[i]; + WorkflowNode node = session.Graph.Nodes.First(n => n.Id == nodeId); + Dictionary inputs = ResolveInputs(session.Graph, node, session.Outputs); + if (!PassEdgeConditions(session.Graph, node, session.Outputs, out string? skipReason)) + { + session.Steps.Add(new NodeRunLog + { + NodeId = node.Id, + Type = node.Type, + Title = node.Title, + Skipped = true, + Message = skipReason, + Inputs = inputs, + }); + continue; + } + + NodeRunLog log = await RunNodeAsync(session.Catalog, node, inputs, cancellationToken); + session.Steps.Add(log); + session.Outputs[node.Id] = log.Outputs; + session.NextIndex = i + 1; + + if (ShouldAskEmptyCities(node, log) && i + 1 < session.Order.Count) + { + return PauseForEmptyCities(session, node, log); + } + } + + return Complete(session); + } + catch (Exception ex) + { + WorkflowRunResult failed = Fail(ex.Message, session.RunId, session.Steps); + session.Completed = true; + session.LastResult = failed; + return failed; + } + } + + private WorkflowRunResult PauseForEmptyCities(WorkflowSession session, WorkflowNode node, NodeRunLog log) + { + string? reason = ReadString(log.Outputs, "reason"); + int timeout = decisionTimeoutSeconds; + if (node.Config.TryGetValue("decisionTimeoutSeconds", out string? configured) + && int.TryParse(configured, out int parsed) + && parsed > 0) + { + timeout = parsed; + } + + DecisionRequest decision = EmptyCityDecision.Create(node.Id, reason, timeout); + session.Pending = decision; + session.TimeoutCts.Cancel(); + session.TimeoutCts.Dispose(); + session.TimeoutCts = new CancellationTokenSource(); + session.LastResult = new WorkflowRunResult + { + Ok = true, + Status = WorkflowRunStatus.NeedsDecision, + RunId = session.RunId, + Decision = decision, + AppliedDecision = session.AppliedDecision, + Steps = session.Steps, + }; + StartTimeout(session); + return session.LastResult; + } + + private void StartTimeout(WorkflowSession session) + { + DecisionRequest? decision = session.Pending; + if (decision is null) + { + return; + } + + CancellationToken token = session.TimeoutCts.Token; + _ = Task.Run(async () => + { + try + { + await Task.Delay(TimeSpan.FromSeconds(decision.TimeoutSeconds), token); + await ResolveAsync( + session, + new DecisionAnswer { OptionId = decision.DefaultOptionId, TimedOut = true }, + CancellationToken.None, + decision.Id); + } + catch (OperationCanceledException) + { + } + catch (Exception ex) + { + session.Completed = true; + session.LastResult = Fail($"默认方案自动确认失败：{ex.Message}", session.RunId, session.Steps); + } + }); + } + + private async Task ResolveAsync( + WorkflowSession session, + DecisionAnswer answer, + CancellationToken cancellationToken, + string? expectedDecisionId) + { + await session.Mutex.WaitAsync(cancellationToken); + try + { + if (session.Completed) + { + return session.LastResult; + } + + if (session.Pending is null + || (!string.IsNullOrEmpty(expectedDecisionId) + && !session.Pending.Id.Equals(expectedDecisionId, StringComparison.OrdinalIgnoreCase))) + { + return session.LastResult; + } + + if (!answer.TimedOut) + { + WorkflowRunResult? invalid = ValidateAnswer(session, answer); + if (invalid is not null) + { + return invalid; + } + } + + session.TimeoutCts.Cancel(); + ApplyAnswer(session, answer); + session.Pending = null; + return await ContinueAsync(session, cancellationToken); + } + catch (Exception ex) + { + return Fail(ex.Message, session.RunId, session.Steps); + } + finally + { + session.Mutex.Release(); + } + } + + private static WorkflowRunResult? ValidateAnswer(WorkflowSession session, DecisionAnswer answer) + { + DecisionRequest decision = session.Pending!; + DecisionOption? option = decision.Options.FirstOrDefault( + item => item.Id.Equals(answer.OptionId, StringComparison.OrdinalIgnoreCase)); + if (option is null) + { + return InvalidDecision(session, "请选择一个列出的方案。"); + } + + if (option.RequiresText && EmptyCityDecision.ParseCities(answer.Text).Count == 0) + { + return InvalidDecision(session, "这个方案需要输入城市名，例如：成都、北京。"); + } + + return null; + } + + private static WorkflowRunResult InvalidDecision(WorkflowSession session, string error) + => new() + { + Ok = false, + Status = WorkflowRunStatus.NeedsDecision, + RunId = session.RunId, + Error = error, + Decision = session.Pending, + AppliedDecision = session.AppliedDecision, + Steps = session.Steps, + }; + + private static void ApplyAnswer(WorkflowSession session, DecisionAnswer answer) + { + DecisionRequest decision = session.Pending + ?? throw new InvalidOperationException("没有待确认的方案。"); + string optionId = string.IsNullOrWhiteSpace(answer.OptionId) + ? decision.DefaultOptionId + : answer.OptionId; + if (answer.TimedOut) + { + optionId = decision.DefaultOptionId; + } + + DecisionOption option = decision.Options.First( + item => item.Id.Equals(optionId, StringComparison.OrdinalIgnoreCase)); + DecisionAnswer applied = new() + { + OptionId = option.Id, + Text = answer.Text, + TimedOut = answer.TimedOut, + }; + session.AppliedDecision = applied; + + NodeRunLog? step = session.Steps.LastOrDefault(item => item.NodeId == decision.NodeId); + if (option.Id == DecisionOptionIds.QueryCities) + { + List cities = EmptyCityDecision.ParseCities(answer.Text); + if (!session.Outputs.TryGetValue(decision.NodeId, out Dictionary? outputs)) + { + outputs = new Dictionary(StringComparer.OrdinalIgnoreCase); + session.Outputs[decision.NodeId] = outputs; + } + + outputs["hasValidCities"] = true; + outputs["cities"] = cities; + outputs["reason"] = $"用户指定城市：{string.Join("、", cities)}"; + if (step is not null) + { + step.Outputs = outputs; + step.Message = Append(step.Message, $"用户确认：改查 {string.Join("、", cities)}。"); + } + } + else if (step is not null) + { + string how = answer.TimedOut ? "超时，自动采用默认方案" : "用户确认默认方案"; + step.Message = Append(step.Message, $"{how}：结束后续查询。"); + } + } + + private static string Append(string? message, string extra) + => string.IsNullOrWhiteSpace(message) ? extra : $"{message}\n{extra}"; + + private static WorkflowRunResult Complete(WorkflowSession session) + { + WorkflowRunResult result = new() + { + Ok = true, + Status = WorkflowRunStatus.Completed, + RunId = session.RunId, + AppliedDecision = session.AppliedDecision, + Steps = session.Steps, + }; + session.Completed = true; + session.Pending = null; + session.LastResult = result; + return result; + } + + private static WorkflowRunResult Fail(string error, string? runId = null, List? steps = null) + => new() + { + Ok = false, + Status = WorkflowRunStatus.Failed, + RunId = runId, + Error = error, + Steps = steps ?? [], + }; + + private static bool ShouldAskEmptyCities(WorkflowNode node, NodeRunLog log) + { + if (log.Skipped) + { + return false; + } + + if (node.Config.TryGetValue("askOnEmptyCities", out string? flag) + && flag.Equals("false", StringComparison.OrdinalIgnoreCase)) + { + return false; + } + + if (!log.Outputs.ContainsKey("hasValidCities")) + { + return false; + } + + return !ReadBool(log.Outputs, "hasValidCities"); + } + + private async Task RunNodeAsync( + NodeCatalogSnapshot snapshot, + WorkflowNode node, + Dictionary inputs, + CancellationToken cancellationToken) + { + LoadedPlugin? plugin = snapshot.LoadedPlugins + .FirstOrDefault(p => p.Manifest.Id.Equals(node.Type, StringComparison.OrdinalIgnoreCase)); + if (plugin is not null) + { + return await RunPluginAsync(plugin, node, inputs, cancellationToken); + } + + AgentTypeInfo? system = AgentCatalog.FindSystem(node.Type) + ?? snapshot.System.FirstOrDefault(item => item.Type.Equals(node.Type, StringComparison.OrdinalIgnoreCase)); + if (system is not null) + { + AgentStepResult step = system.Handler switch + { + SystemHandler.FileCity => await FileCityAgent.RunAsync(fileCityAgent, inputs, cancellationToken), + SystemHandler.Weather => await WeatherAgent.RunAsync(weatherAgent, inputs, cancellationToken), + _ => throw new InvalidOperationException( + $"系统节点 `{system.Type}` 在 AgentCatalog 里没有绑定实现。请给它设置 Handler。"), + }; + return ToLog(node, step); + } + + throw new InvalidOperationException( + $"未找到节点类型 `{node.Type}`。系统节点来自 AgentCatalog，插件节点来自 plugins 目录，请点「刷新节点」。"); + } + + private async Task RunPluginAsync( + LoadedPlugin plugin, + WorkflowNode node, + Dictionary inputs, + CancellationToken cancellationToken) + { + Dictionary declared = FilterDeclaredInputs(plugin.Manifest, inputs); + node.Config.TryGetValue("credentialId", out string? credentialId); + Dictionary outputs = await plugins.RunAsync(plugin, declared, credentialId, cancellationToken); + string? stderr = null; + if (outputs.Remove("_stderr", out object? stderrValue)) + { + stderr = stderrValue?.ToString(); + } + + return new NodeRunLog + { + NodeId = node.Id, + Type = node.Type, + Title = node.Title, + Message = stderr, + Inputs = declared, + Outputs = outputs, + }; + } + + private static NodeRunLog ToLog(WorkflowNode node, AgentStepResult step) + => new() + { + NodeId = node.Id, + Type = node.Type, + Title = node.Title, + Message = step.Message, + Inputs = step.Inputs, + Outputs = step.Outputs, + }; + + private static Dictionary FilterDeclaredInputs(PluginManifest manifest, Dictionary inputs) + { + Dictionary declared = new(StringComparer.OrdinalIgnoreCase); + foreach (PluginPort port in manifest.Inputs) + { + if (inputs.TryGetValue(port.Name, out object? value)) + { + declared[port.Name] = value; + } + } + + return declared; + } + + private static void Validate(WorkflowGraph graph) + { + if (graph.Nodes.Count == 0) + { + throw new InvalidOperationException("画布上还没有节点。"); + } + + HashSet ids = graph.Nodes.Select(n => n.Id).ToHashSet(); + foreach (WorkflowEdge edge in graph.Edges) + { + if (!ids.Contains(edge.From) || !ids.Contains(edge.To)) + { + throw new InvalidOperationException("存在指向已删除节点的连线。"); + } + } + } + + private static List TopologicalOrder(WorkflowGraph graph) + { + Dictionary indegree = graph.Nodes.ToDictionary(n => n.Id, _ => 0); + Dictionary> outgoing = graph.Nodes.ToDictionary(n => n.Id, _ => new List()); + foreach (WorkflowEdge edge in graph.Edges) + { + indegree[edge.To]++; + outgoing[edge.From].Add(edge.To); + } + + Queue ready = new(indegree.Where(kv => kv.Value == 0).Select(kv => kv.Key)); + List order = []; + while (ready.Count > 0) + { + string id = ready.Dequeue(); + order.Add(id); + foreach (string next in outgoing[id].Distinct()) + { + indegree[next]--; + if (indegree[next] == 0) + { + ready.Enqueue(next); + } + } + } + + if (order.Count != graph.Nodes.Count) + { + throw new InvalidOperationException("工作流存在环，无法运行。"); + } + + return order; + } + + private static Dictionary ResolveInputs( + WorkflowGraph graph, + WorkflowNode node, + Dictionary> outputs) + { + Dictionary inputs = new(StringComparer.OrdinalIgnoreCase); + foreach (KeyValuePair item in node.Config) + { + if (ReservedConfigKeys.Contains(item.Key) || string.IsNullOrWhiteSpace(item.Value)) + { + continue; + } + + inputs[item.Key] = item.Value; + } + + foreach (WorkflowEdge edge in graph.Edges.Where(e => e.To == node.Id)) + { + if (!outputs.TryGetValue(edge.From, out Dictionary? source)) + { + continue; + } + + if (source.TryGetValue(edge.FromPort, out object? value)) + { + inputs[edge.ToPort] = value; + } + } + + return inputs; + } + + private static bool PassEdgeConditions( + WorkflowGraph graph, + WorkflowNode node, + Dictionary> outputs, + out string? skipReason) + { + List incoming = graph.Edges.Where(e => e.To == node.Id).ToList(); + if (incoming.Count == 0) + { + skipReason = null; + return true; + } + + foreach (WorkflowEdge edge in incoming) + { + if (string.IsNullOrWhiteSpace(edge.When)) + { + continue; + } + + if (!outputs.TryGetValue(edge.From, out Dictionary? source)) + { + skipReason = $"上一节点 {edge.From} 尚未产出结果。"; + return false; + } + + bool hasCities = ReadBool(source, "hasValidCities"); + bool pass = edge.When switch + { + "hasValidCities" => hasCities, + "!hasValidCities" => !hasCities, + _ => true, + }; + if (!pass) + { + skipReason = $"连线条件 {edge.When} 不满足，跳过本节点。"; + return false; + } + } + + skipReason = null; + return true; + } + + private static bool ReadBool(Dictionary map, string key) + { + if (!map.TryGetValue(key, out object? value) || value is null) + { + return false; + } + + return value switch + { + bool b => b, + JsonElement el when el.ValueKind == JsonValueKind.True => true, + JsonElement el when el.ValueKind == JsonValueKind.False => false, + _ => bool.TryParse(value.ToString(), out bool parsed) && parsed, + }; + } + + private static string? ReadString(Dictionary map, string key) + { + if (!map.TryGetValue(key, out object? value) || value is null) + { + return null; + } + + if (value is JsonElement el) + { + return el.ValueKind == JsonValueKind.String ? el.GetString() : el.ToString(); + } + + return value.ToString(); + } +} diff --git a/Web/WorkflowModels.cs b/MAF1.Web/Web/WorkflowModels.cs similarity index 75% rename from Web/WorkflowModels.cs rename to MAF1.Web/Web/WorkflowModels.cs index 362ff41..67eb418 100644 --- a/Web/WorkflowModels.cs +++ b/MAF1.Web/Web/WorkflowModels.cs @@ -1,3 +1,5 @@ +using MAF1.Decisions; + namespace MAF1.Web; public sealed class WorkflowGraph @@ -26,10 +28,21 @@ public sealed class WorkflowEdge public string? When { get; set; } } +public static class WorkflowRunStatus +{ + public const string Completed = "completed"; + public const string NeedsDecision = "needsDecision"; + public const string Failed = "failed"; +} + public sealed class WorkflowRunResult { public bool Ok { get; set; } + public string Status { get; set; } = WorkflowRunStatus.Completed; + public string? RunId { get; set; } public string? Error { get; set; } + public DecisionRequest? Decision { get; set; } + public DecisionAnswer? AppliedDecision { get; set; } public List Steps { get; set; } = []; } diff --git a/MAF1.Web/Web/WorkflowRunStore.cs b/MAF1.Web/Web/WorkflowRunStore.cs new file mode 100644 index 0000000..2b83ea6 --- /dev/null +++ b/MAF1.Web/Web/WorkflowRunStore.cs @@ -0,0 +1,35 @@ +using System.Collections.Concurrent; +using MAF1.Decisions; +using MAF1.Plugins; + +namespace MAF1.Web; + +internal sealed class WorkflowSession +{ + public required string RunId { get; init; } + public required WorkflowGraph Graph { get; init; } + public required NodeCatalogSnapshot Catalog { get; init; } + public required List Order { get; init; } + public required Dictionary> Outputs { get; init; } + public required List Steps { get; init; } + public int NextIndex { get; set; } + public DecisionRequest? Pending { get; set; } + public bool Completed { get; set; } + public WorkflowRunResult LastResult { get; set; } = new(); + public DecisionAnswer? AppliedDecision { get; set; } + public CancellationTokenSource TimeoutCts { get; set; } = new(); + public SemaphoreSlim Mutex { get; } = new(1, 1); + public object Sync { get; } = new(); +} + +public sealed class WorkflowRunStore +{ + private readonly ConcurrentDictionary _sessions = new(StringComparer.OrdinalIgnoreCase); + + internal bool TryAdd(WorkflowSession session) => _sessions.TryAdd(session.RunId, session); + + internal bool TryGet(string runId, out WorkflowSession? session) => _sessions.TryGetValue(runId, out session); + + public WorkflowRunResult? GetResult(string runId) + => _sessions.TryGetValue(runId, out WorkflowSession? session) ? session.LastResult : null; +} diff --git a/appsettings.json b/MAF1.Web/appsettings.json similarity index 92% rename from appsettings.json rename to MAF1.Web/appsettings.json index 9385819..875d277 100644 --- a/appsettings.json +++ b/MAF1.Web/appsettings.json @@ -4,6 +4,9 @@ "Endpoint": "https://api.deepseek.com", "Model": "deepseek-v4-flash" }, + "Workflow": { + "DecisionTimeoutSeconds": 30 + }, "Plugins": { "Directory": "plugins", "DefaultTimeoutSeconds": 180 diff --git a/wwwroot/css/app.css b/MAF1.Web/wwwroot/css/app.css similarity index 71% rename from wwwroot/css/app.css rename to MAF1.Web/wwwroot/css/app.css index 9ff88e4..14d0f62 100644 --- a/wwwroot/css/app.css +++ b/MAF1.Web/wwwroot/css/app.css @@ -243,3 +243,93 @@ h2 { .wire.selected { stroke: var(--weather); } + +.modal { + position: fixed; + inset: 0; + background: rgba(6, 10, 16, 0.72); + display: flex; + align-items: center; + justify-content: center; + z-index: 20; + padding: 20px; +} + + .modal.hidden { + display: none; + } + +.dialog { + width: min(480px, 100%); + background: var(--panel); + border: 1px solid var(--line); + border-radius: 12px; + padding: 18px 18px 16px; +} + + .dialog h2 { + margin: 0 0 8px; + font-size: 16px; + } + + .dialog p { + margin: 0 0 8px; + font-size: 13px; + line-height: 1.5; + } + +.decision-options { + display: flex; + flex-direction: column; + gap: 8px; + margin: 12px 0; +} + +.decision-option { + display: block; + background: #10161d; + border: 1px solid var(--line); + border-radius: 8px; + padding: 10px 12px; + cursor: pointer; +} + + .decision-option input { + margin-right: 8px; + } + + .decision-option.selected { + border-color: var(--accent); + } + + .decision-option strong { + display: block; + font-size: 13px; + } + + .decision-option span { + display: block; + margin-top: 4px; + color: var(--muted); + font-size: 12px; + } + +.dialog input { + width: 100%; + background: #10161d; + color: var(--text); + border: 1px solid var(--line); + border-radius: 6px; + padding: 8px; + margin-bottom: 8px; +} + +.dialog-actions { + display: flex; + justify-content: flex-end; + margin-top: 8px; +} + +.hidden { + display: none !important; +} diff --git a/wwwroot/index.html b/MAF1.Web/wwwroot/index.html similarity index 64% rename from wwwroot/index.html rename to MAF1.Web/wwwroot/index.html index ea62ffb..704ac62 100644 --- a/wwwroot/index.html +++ b/MAF1.Web/wwwroot/index.html @@ -37,6 +37,21 @@

运行结果

尚未运行。
+ diff --git a/wwwroot/js/app.js b/MAF1.Web/wwwroot/js/app.js similarity index 69% rename from wwwroot/js/app.js rename to MAF1.Web/wwwroot/js/app.js index 5076cff..b91114b 100644 --- a/wwwroot/js/app.js +++ b/MAF1.Web/wwwroot/js/app.js @@ -15,6 +15,20 @@ const canvas = document.getElementById("canvas"); const wires = document.getElementById("wires"); const inspector = document.getElementById("inspector"); const logEl = document.getElementById("log"); +const decisionModal = document.getElementById("decisionModal"); +const decisionPrompt = document.getElementById("decisionPrompt"); +const decisionReason = document.getElementById("decisionReason"); +const decisionRemain = document.getElementById("decisionRemain"); +const decisionOptions = document.getElementById("decisionOptions"); +const decisionText = document.getElementById("decisionText"); +const decisionTextLabel = document.getElementById("decisionTextLabel"); +const decisionError = document.getElementById("decisionError"); +const decisionSubmit = document.getElementById("decisionSubmit"); +const btnRun = document.getElementById("btnRun"); + +let decisionTimer = null; +let pendingRunId = null; +let selectedOptionId = ""; function uid(prefix) { return prefix + Math.random().toString(36).slice(2, 8); @@ -280,25 +294,159 @@ function renderInspector() { } async function runGraph() { + closeDecision(); logEl.textContent = "运行中…"; - const res = await fetch("/api/run", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ nodes: state.nodes, edges: state.edges }), - }); - const data = await res.json(); - if (!data.ok) { - logEl.textContent = data.error ?? "运行失败"; - return; + btnRun.disabled = true; + try { + const res = await fetch("/api/run", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ nodes: state.nodes, edges: state.edges }), + }); + const data = await res.json(); + handleRunResult(data); + } catch (err) { + logEl.textContent = err?.message ?? "运行失败"; + btnRun.disabled = false; } - logEl.textContent = data.steps.map((step) => { +} + +function formatRunLog(data) { + const applied = data.appliedDecision + ? `\n已选方案: ${data.appliedDecision.optionId}${data.appliedDecision.timedOut ? "（超时默认）" : ""}${data.appliedDecision.text ? " / " + data.appliedDecision.text : ""}\n` + : ""; + const steps = (data.steps || []).map((step) => { const head = `${step.title} (${step.type})${step.skipped ? " [跳过]" : ""}`; const inputs = JSON.stringify(step.inputs, null, 2); const outputs = JSON.stringify(step.outputs, null, 2); return `${head}\n输入:\n${inputs}\n输出:\n${outputs}\n${step.message ?? ""}`; }).join("\n\n-----\n\n"); + return (applied + (steps || "没有步骤。")).trim(); } +function handleRunResult(data) { + if (data.status === "needsDecision" && data.decision) { + logEl.textContent = "等待确认…\n\n" + formatRunLog(data); + showDecision(data, data.error); + return; + } + closeDecision(); + btnRun.disabled = false; + if (!data.ok) { + logEl.textContent = data.error ?? "运行失败"; + return; + } + logEl.textContent = formatRunLog(data); +} + +function showDecision(data, errorText) { + const decision = data.decision; + pendingRunId = data.runId; + selectedOptionId = decision.defaultOptionId; + decisionPrompt.textContent = decision.prompt || "请选择一个方案。"; + decisionReason.textContent = decision.reason ? `原因：${decision.reason}` : ""; + decisionError.textContent = errorText || ""; + decisionError.classList.toggle("hidden", !errorText); + decisionOptions.innerHTML = (decision.options || []).map((option) => ` + `).join(""); + decisionOptions.querySelectorAll(".decision-option").forEach((el) => { + el.onchange = () => { + selectedOptionId = el.dataset.id; + decisionOptions.querySelectorAll(".decision-option").forEach((item) => { + item.classList.toggle("selected", item.dataset.id === selectedOptionId); + }); + syncDecisionText(); + }; + }); + decisionText.placeholder = (decision.options || []).find((o) => o.requiresText)?.textPlaceholder || "请输入"; + syncDecisionText(); + decisionModal.classList.remove("hidden"); + decisionModal.setAttribute("aria-hidden", "false"); + startDecisionTimer(decision.deadline); +} + +function syncDecisionText() { + const option = decisionOptions.querySelector(`.decision-option[data-id="${selectedOptionId}"]`); + const needsText = option?.dataset.requires === "1"; + decisionText.classList.toggle("hidden", !needsText); + decisionTextLabel.classList.toggle("hidden", !needsText); + if (needsText) { + decisionText.focus(); + } +} + +function startDecisionTimer(deadline) { + if (decisionTimer) { + clearInterval(decisionTimer); + } + const tick = async () => { + const remain = Math.max(0, Math.ceil((new Date(deadline).getTime() - Date.now()) / 1000)); + decisionRemain.textContent = String(remain); + if (remain <= 0) { + clearInterval(decisionTimer); + decisionTimer = null; + if (pendingRunId) { + await pollRun(pendingRunId); + } + } + }; + tick(); + decisionTimer = setInterval(tick, 500); +} + +async function pollRun(runId) { + for (let i = 0; i < 8; i++) { + try { + const res = await fetch(`/api/run/${runId}`); + const data = await res.json(); + if (data.status !== "needsDecision") { + handleRunResult(data); + return; + } + } catch { + btnRun.disabled = false; + return; + } + await new Promise((resolve) => setTimeout(resolve, 250)); + } +} + +function closeDecision() { + if (decisionTimer) { + clearInterval(decisionTimer); + decisionTimer = null; + } + pendingRunId = null; + decisionModal.classList.add("hidden"); + decisionModal.setAttribute("aria-hidden", "true"); +} + +decisionSubmit.onclick = async () => { + if (!pendingRunId || !selectedOptionId) { + return; + } + decisionSubmit.disabled = true; + decisionError.classList.add("hidden"); + try { + const res = await fetch(`/api/run/${pendingRunId}/decide`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ optionId: selectedOptionId, text: decisionText.value }), + }); + const data = await res.json(); + handleRunResult(data); + } catch (err) { + decisionError.textContent = err?.message ?? "确认失败"; + decisionError.classList.remove("hidden"); + } finally { + decisionSubmit.disabled = false; + } +}; + async function loadCatalog() { const [catalog, credentials] = await Promise.all([ (await fetch("/api/catalog")).json(), @@ -323,7 +471,7 @@ document.getElementById("btnExample").onclick = () => { state.selected = { kind: "edge", id: "e1" }; render(); }; -document.getElementById("btnRun").onclick = runGraph; +btnRun.onclick = runGraph; document.getElementById("btnRefresh").onclick = async () => { logEl.textContent = "正在重新扫描系统节点和 plugins 目录…"; await loadCatalog(); diff --git a/MAF1.slnx b/MAF1.slnx index 57bcbb0..1894f29 100644 --- a/MAF1.slnx +++ b/MAF1.slnx @@ -7,5 +7,6 @@ - + + diff --git a/CliHost.cs b/MAF1/CliHost.cs similarity index 79% rename from CliHost.cs rename to MAF1/CliHost.cs index f3a1217..da1c598 100644 --- a/CliHost.cs +++ b/MAF1/CliHost.cs @@ -33,13 +33,15 @@ internal static class CliHost return; } + int decisionTimeoutSeconds = Math.Clamp(config.GetValue("Workflow:DecisionTimeoutSeconds", 30), 1, 600); Console.WriteLine($"天气数据源: {weatherOptions.Provider}"); Console.WriteLine($"目标文件: {filePath}"); + Console.WriteLine($"决策等待: {decisionTimeoutSeconds} 秒（超时采用默认方案：结束查询）"); Console.WriteLine(); Workflow workflow = mode == "edge" - ? CityWeatherWorkflow.BuildEdgeCondition(fileCityAgent, weatherAgent) - : CityWeatherWorkflow.BuildNodeCondition(fileCityAgent, weatherAgent); + ? CityWeatherWorkflow.BuildEdgeCondition(fileCityAgent, weatherAgent, decisionTimeoutSeconds) + : CityWeatherWorkflow.BuildNodeCondition(fileCityAgent, weatherAgent, decisionTimeoutSeconds); string label = mode == "edge" ? "模式 edge：判断写在边上" : "模式 node：判断写在节点里"; @@ -50,6 +52,11 @@ internal static class CliHost { mode = "node"; filePath = Path.Combine("Data", "cities.txt"); + if (args.Length == 0) + { + return true; + } + IEnumerable rest = args[0] is "--cli" ? args.Skip(1) : args; string[] list = rest.ToArray(); if (list.Length == 0) @@ -82,12 +89,13 @@ internal static class CliHost { Console.WriteLine( """ - 可视化编排（默认）: - dotnet run + 命令行工作流: + dotnet run --project MAF1 + dotnet run --project MAF1 -- node Data/cities.txt + dotnet run --project MAF1 -- edge Data/cities.txt - 命令行工作流对比: - dotnet run -- node Data/cities.txt - dotnet run -- edge Data/cities.txt + 网页编排在另一个项目: + dotnet run --project MAF1.Web """); } } diff --git a/MAF1/Data/cities.txt b/MAF1/Data/cities.txt new file mode 100644 index 0000000..3233868 --- /dev/null +++ b/MAF1/Data/cities.txt @@ -0,0 +1,3 @@ +成都 +阿姆斯特丹 +北京 diff --git a/MAF1/Data/not-cities.txt b/MAF1/Data/not-cities.txt new file mode 100644 index 0000000..dcd167e --- /dev/null +++ b/MAF1/Data/not-cities.txt @@ -0,0 +1,3 @@ +香蕉 +hello +不是城市 diff --git a/MAF1/MAF1.csproj b/MAF1/MAF1.csproj new file mode 100644 index 0000000..2e982c5 --- /dev/null +++ b/MAF1/MAF1.csproj @@ -0,0 +1,32 @@ + + + + Exe + net10.0 + enable + enable + MAF1 + app.manifest + + + + + + + + + + + + + + + + PreserveNewest + + + PreserveNewest + + + + diff --git a/MAF1/Orchestration/ConsoleDecisionPrompt.cs b/MAF1/Orchestration/ConsoleDecisionPrompt.cs new file mode 100644 index 0000000..22cac8a --- /dev/null +++ b/MAF1/Orchestration/ConsoleDecisionPrompt.cs @@ -0,0 +1,91 @@ +using MAF1.Decisions; + +namespace MAF1.Orchestration; + +internal static class ConsoleDecisionPrompt +{ + public static async Task WaitAsync(DecisionRequest request) + { + Console.WriteLine(); + Console.WriteLine("[需要确认]"); + Console.WriteLine(request.Prompt); + if (!string.IsNullOrWhiteSpace(request.Reason)) + { + Console.WriteLine($"原因：{request.Reason}"); + } + + Console.WriteLine($"未及时确认将使用默认方案。等待 {request.TimeoutSeconds} 秒。"); + for (int i = 0; i < request.Options.Count; i++) + { + DecisionOption option = request.Options[i]; + string mark = option.IsDefault ? " ← 默认" : ""; + Console.WriteLine($" {i + 1}) {option.Label}{mark}"); + if (!string.IsNullOrWhiteSpace(option.Description)) + { + Console.WriteLine($" {option.Description}"); + } + } + + Console.Write("输入序号，或直接回车采用默认："); + string? line = await ReadLineOrTimeoutAsync(request.Deadline); + if (line is null) + { + Console.WriteLine(); + Console.WriteLine("未及时确认，采用默认方案：结束查询。"); + return new DecisionAnswer { OptionId = request.DefaultOptionId, TimedOut = true }; + } + + line = line.Trim(); + if (line.Length == 0 || line is "1" || line.Equals("stop", StringComparison.OrdinalIgnoreCase)) + { + return new DecisionAnswer { OptionId = DecisionOptionIds.Stop }; + } + + if (line is "2" || line.Equals(DecisionOptionIds.QueryCities, StringComparison.OrdinalIgnoreCase)) + { + Console.Write("请输入城市名（例如 成都、北京）："); + string? cities = await ReadLineOrTimeoutAsync(request.Deadline); + if (cities is null) + { + Console.WriteLine(); + Console.WriteLine("未及时输入城市，采用默认方案：结束查询。"); + return new DecisionAnswer { OptionId = DecisionOptionIds.Stop, TimedOut = true }; + } + + if (EmptyCityDecision.ParseCities(cities).Count == 0) + { + Console.WriteLine("没有有效城市名，采用默认方案：结束查询。"); + return new DecisionAnswer { OptionId = DecisionOptionIds.Stop }; + } + + return new DecisionAnswer { OptionId = DecisionOptionIds.QueryCities, Text = cities }; + } + + if (EmptyCityDecision.ParseCities(line).Count > 0) + { + return new DecisionAnswer { OptionId = DecisionOptionIds.QueryCities, Text = line }; + } + + Console.WriteLine("无法识别输入，采用默认方案：结束查询。"); + return new DecisionAnswer { OptionId = DecisionOptionIds.Stop }; + } + + private static async Task ReadLineOrTimeoutAsync(DateTimeOffset deadline) + { + TimeSpan remain = deadline - DateTimeOffset.UtcNow; + if (remain <= TimeSpan.Zero) + { + return null; + } + + Task read = Task.Run(() => Console.ReadLine()); + Task delay = Task.Delay(remain); + Task finished = await Task.WhenAny(read, delay); + if (finished == delay) + { + return null; + } + + return await read; + } +} diff --git a/Orchestration/WorkflowOrchestration.cs b/MAF1/Orchestration/WorkflowOrchestration.cs similarity index 70% rename from Orchestration/WorkflowOrchestration.cs rename to MAF1/Orchestration/WorkflowOrchestration.cs index 9ad3df4..4ca434d 100644 --- a/Orchestration/WorkflowOrchestration.cs +++ b/MAF1/Orchestration/WorkflowOrchestration.cs @@ -1,3 +1,4 @@ +using MAF1.Decisions; using Microsoft.Agents.AI; using Microsoft.Agents.AI.Workflows; @@ -19,6 +20,12 @@ public static class WorkflowOrchestration { switch (evt) { + case RequestInfoEvent requestEvent: + DecisionRequest decision = ReadDecision(requestEvent.Request); + DecisionAnswer answer = await ConsoleDecisionPrompt.WaitAsync(decision); + await run.SendResponseAsync(requestEvent.Request.CreateResponse(answer)); + break; + case AgentResponseUpdateEvent update: if (update.ExecutorId != lastExecutorId) { @@ -47,4 +54,14 @@ public static class WorkflowOrchestration Console.WriteLine(); } + + private static DecisionRequest ReadDecision(ExternalRequest request) + { + if (request.TryGetDataAs(out DecisionRequest? decision) && decision is not null) + { + return decision; + } + + throw new InvalidOperationException("工作流发出了无法识别的确认请求。"); + } } diff --git a/MAF1/Program.cs b/MAF1/Program.cs new file mode 100644 index 0000000..f0ce8ca --- /dev/null +++ b/MAF1/Program.cs @@ -0,0 +1,5 @@ +using MAF1; +using MAF1.Utils; + +WindowsConsole.EnableUtf8(); +await CliHost.RunAsync(args); diff --git a/Properties/launchSettings.json b/MAF1/Properties/launchSettings.json similarity index 58% rename from Properties/launchSettings.json rename to MAF1/Properties/launchSettings.json index 81fbea7..0a4bb50 100644 --- a/Properties/launchSettings.json +++ b/MAF1/Properties/launchSettings.json @@ -1,15 +1,5 @@ { "profiles": { - "designer": { - "commandName": "Project", - "applicationUrl": "http://127.0.0.1:5288", - "environmentVariables": { - "ASPNETCORE_URLS": "http://127.0.0.1:5288", - "OPENAI_ENDPOINT": "https://api.deepseek.com", - "OPENAI_API_KEY": "sk-b31330cf3a414d86ab614b9dfb3650c3", - "OPENAI_CHAT_MODEL": "deepseek-v4-flash" - } - }, "node": { "commandName": "Project", "commandLineArgs": "node Data/cities.txt", @@ -21,7 +11,7 @@ }, "edge": { "commandName": "Project", - "commandLineArgs": "edge Data/cities.txt", + "commandLineArgs": "edge Data/not-cities.txt", "environmentVariables": { "OPENAI_ENDPOINT": "https://api.deepseek.com", "OPENAI_API_KEY": "sk-b31330cf3a414d86ab614b9dfb3650c3", diff --git a/MAF1/Workflows/ApplyCityDecisionExecutor.cs b/MAF1/Workflows/ApplyCityDecisionExecutor.cs new file mode 100644 index 0000000..e927548 --- /dev/null +++ b/MAF1/Workflows/ApplyCityDecisionExecutor.cs @@ -0,0 +1,42 @@ +using MAF1.Agents.FileCity; +using MAF1.Decisions; +using Microsoft.Agents.AI.Workflows; + +namespace MAF1.Workflows; + +/// +/// 把控制台确认结果变回工作流消息：继续则送出城市，默认/超时则结束。 +/// +internal sealed class ApplyCityDecisionExecutor() : Executor("ApplyCityDecision") +{ + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder builder) + { + return base.ConfigureProtocol(builder) + .SendsMessage() + .YieldsOutput(); + } + + public override async ValueTask HandleAsync( + DecisionAnswer answer, + IWorkflowContext context, + CancellationToken cancellationToken = default) + { + if (EmptyCityDecision.TryContinueWithCities(answer, out List cities)) + { + await context.SendMessageAsync( + new CityExtraction + { + HasValidCities = true, + Cities = cities, + Reason = $"用户指定城市：{string.Join("、", cities)}", + }, + cancellationToken: cancellationToken); + return; + } + + string how = answer.TimedOut ? "超时，自动采用默认方案" : "已确认默认方案"; + await context.YieldOutputAsync( + $"{how}：结束后续查询，不执行 WeatherAgent。", + cancellationToken); + } +} diff --git a/MAF1/Workflows/AskCityDecisionExecutor.cs b/MAF1/Workflows/AskCityDecisionExecutor.cs new file mode 100644 index 0000000..49ee289 --- /dev/null +++ b/MAF1/Workflows/AskCityDecisionExecutor.cs @@ -0,0 +1,26 @@ +using MAF1.Agents.FileCity; +using MAF1.Decisions; +using Microsoft.Agents.AI.Workflows; + +namespace MAF1.Workflows; + +/// +/// 边条件模式：没有城市时不在边上直接结束，而是发出决策请求。 +/// +internal sealed class AskCityDecisionExecutor(int timeoutSeconds) : Executor("AskCityDecision") +{ + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder builder) + { + return base.ConfigureProtocol(builder).SendsMessage(); + } + + public override async ValueTask HandleAsync( + CityExtraction extraction, + IWorkflowContext context, + CancellationToken cancellationToken = default) + { + await context.SendMessageAsync( + EmptyCityDecision.Create("AskCityDecision", extraction.Reason, timeoutSeconds), + cancellationToken: cancellationToken); + } +} diff --git a/Workflows/CityGateExecutor.cs b/MAF1/Workflows/CityGateExecutor.cs similarity index 77% rename from Workflows/CityGateExecutor.cs rename to MAF1/Workflows/CityGateExecutor.cs index 6bbcab4..a9d8a8f 100644 --- a/Workflows/CityGateExecutor.cs +++ b/MAF1/Workflows/CityGateExecutor.cs @@ -1,10 +1,11 @@ using MAF1.Agents.FileCity; +using MAF1.Decisions; using Microsoft.Agents.AI.Workflows; using Microsoft.Extensions.AI; namespace MAF1.Workflows; -internal sealed class CityGateExecutor() : ChatProtocolExecutor( +internal sealed class CityGateExecutor(int decisionTimeoutSeconds) : ChatProtocolExecutor( "CityGate", new ChatProtocolExecutorOptions { AutoSendTurnToken = false }) { @@ -13,6 +14,7 @@ internal sealed class CityGateExecutor() : ChatProtocolExecutor( return base.ConfigureProtocol(builder) .SendsMessage() .SendsMessage() + .SendsMessage() .YieldsOutput(); } @@ -31,12 +33,9 @@ internal sealed class CityGateExecutor() : ChatProtocolExecutor( CityExtraction extraction = FileCityAgent.Parse(text); if (!extraction.HasValidCities) { - string reason = string.IsNullOrWhiteSpace(extraction.Reason) - ? "文件里没有可查询的城市。" - : extraction.Reason; - await context.YieldOutputAsync( - $"没有有效城市，工作流结束，不执行 WeatherAgent。{reason}", - cancellationToken); + await context.SendMessageAsync( + EmptyCityDecision.Create("CityGate", extraction.Reason, decisionTimeoutSeconds), + cancellationToken: cancellationToken); return; } diff --git a/Workflows/CityParseExecutor.cs b/MAF1/Workflows/CityParseExecutor.cs similarity index 100% rename from Workflows/CityParseExecutor.cs rename to MAF1/Workflows/CityParseExecutor.cs diff --git a/Workflows/CityWeatherWorkflow.cs b/MAF1/Workflows/CityWeatherWorkflow.cs similarity index 56% rename from Workflows/CityWeatherWorkflow.cs rename to MAF1/Workflows/CityWeatherWorkflow.cs index 8ffdfec..2b736d6 100644 --- a/Workflows/CityWeatherWorkflow.cs +++ b/MAF1/Workflows/CityWeatherWorkflow.cs @@ -1,3 +1,4 @@ +using MAF1.Decisions; using Microsoft.Agents.AI; using Microsoft.Agents.AI.Workflows; using MAF1.Agents.FileCity; @@ -6,32 +7,44 @@ namespace MAF1.Workflows; public static class CityWeatherWorkflow { - public static Workflow BuildNodeCondition(AIAgent fileCityAgent, AIAgent weatherAgent) + public static Workflow BuildNodeCondition(AIAgent fileCityAgent, AIAgent weatherAgent, int decisionTimeoutSeconds) { (ExecutorBinding fileCity, ExecutorBinding weather) = BindAgents(fileCityAgent, weatherAgent); - CityGateExecutor gate = new(); + CityGateExecutor gate = new(decisionTimeoutSeconds); + RequestPort cityDecision = RequestPort.Create("CityDecision"); + ApplyCityDecisionExecutor apply = new(); + ToWeatherPromptExecutor toWeather = new(); return new WorkflowBuilder(fileCity) .AddEdge(fileCity, gate) .AddEdge(gate, weather) - .WithOutputFrom(gate, weather) + .AddEdge(gate, cityDecision) + .AddEdge(cityDecision, apply) + .AddEdge(apply, toWeather, extraction => extraction is { HasValidCities: true }) + .AddEdge(toWeather, weather) + .WithOutputFrom(apply, weather) .WithName("CityWeather-Node") .Build(); } - public static Workflow BuildEdgeCondition(AIAgent fileCityAgent, AIAgent weatherAgent) + public static Workflow BuildEdgeCondition(AIAgent fileCityAgent, AIAgent weatherAgent, int decisionTimeoutSeconds) { (ExecutorBinding fileCity, ExecutorBinding weather) = BindAgents(fileCityAgent, weatherAgent); CityParseExecutor parse = new(); ToWeatherPromptExecutor toWeather = new(); - SkipWeatherExecutor skip = new(); + AskCityDecisionExecutor ask = new(decisionTimeoutSeconds); + RequestPort cityDecision = RequestPort.Create("CityDecision"); + ApplyCityDecisionExecutor apply = new(); return new WorkflowBuilder(fileCity) .AddEdge(fileCity, parse) .AddEdge(parse, toWeather, extraction => extraction is { HasValidCities: true }) - .AddEdge(parse, skip, extraction => extraction is not { HasValidCities: true }) + .AddEdge(parse, ask, extraction => extraction is not { HasValidCities: true }) + .AddEdge(ask, cityDecision) + .AddEdge(cityDecision, apply) + .AddEdge(apply, toWeather, extraction => extraction is { HasValidCities: true }) .AddEdge(toWeather, weather) - .WithOutputFrom(skip, weather) + .WithOutputFrom(apply, weather) .WithName("CityWeather-Edge") .Build(); } diff --git a/Workflows/ToWeatherPromptExecutor.cs b/MAF1/Workflows/ToWeatherPromptExecutor.cs similarity index 100% rename from Workflows/ToWeatherPromptExecutor.cs rename to MAF1/Workflows/ToWeatherPromptExecutor.cs diff --git a/app.manifest b/MAF1/app.manifest similarity index 100% rename from app.manifest rename to MAF1/app.manifest diff --git a/MAF1/appsettings.json b/MAF1/appsettings.json new file mode 100644 index 0000000..59c0d5e --- /dev/null +++ b/MAF1/appsettings.json @@ -0,0 +1,21 @@ +{ + "Llm": { + "ApiKey": "", + "Endpoint": "https://api.deepseek.com", + "Model": "deepseek-v4-flash" + }, + "Workflow": { + "DecisionTimeoutSeconds": 30 + }, + "Weather": { + "Provider": "Wttr", + "Language": "zh", + "Wttr": { + "UrlTemplate": "https://wttr.in/{location}?lang={lang}&format=3" + }, + "OpenWeather": { + "UrlTemplate": "https://api.openweathermap.org/data/2.5/weather?q={location}&appid={apiKey}&units=metric&lang={lang}", + "ApiKey": "" + } + } +} diff --git a/Plugins/README.md b/Plugins/README.md index f05e5cc..aa583db 100644 --- a/Plugins/README.md +++ b/Plugins/README.md @@ -1,6 +1,6 @@ # 插件目录 -每个子文件夹是一个插件。软件启动后扫描**执行目录**下的 `plugins/`（即 `MAF1.exe` 旁边），不是源码目录。 +每个子文件夹是一个插件。软件启动后扫描**网页宿主执行目录**下的 `plugins/`（即 `MAF1.Web.exe` 旁边），不是源码目录。 ## 必备文件 diff --git a/README.md b/README.md index c08b820..3295a72 100644 --- a/README.md +++ b/README.md @@ -1,13 +1,13 @@ # MAF1 — 多智能体工作流编排 -MAF1 是一个基于 **.NET 10** 和 **Microsoft Agents AI** 的多智能体（Multi-Agent）演示项目。它把「读文件抽城市 → 查天气」这条业务链路做成两种用法： +MAF1 是一个基于 **.NET 10** 和 **Microsoft Agents AI** 的多智能体演示仓库。同一条「读文件抽城市 → 查天气」链路拆成两个可执行项目： -1. **可视化编排器**：浏览器里拖节点、连线、配条件，然后一键运行。 -2. **命令行工作流**：用 Microsoft Agents AI Workflows 对比「条件写在节点里」和「条件写在边上」两种编排方式。 +1. **MAF1**：命令行工作流。用 Microsoft Agents AI Workflows 对比「条件写在节点里」和「条件写在边上」。 +2. **MAF1.Web**：可视化编排器。浏览器里拖节点、连线、配条件，然后一键运行。 -内置 Agent 在进程内执行；同一套能力也可以以 **独立进程插件** 的形式出现在画布上。LLM 的 endpoint / API Key / 模型由宿主注入，**不会**画成节点端口，也不会写进流程图 JSON。 +内置 Agent 在 `MAF1.Core` 里；网页还可以把同一套能力以 **独立进程插件** 的形式画到画布上。LLM 的 endpoint / API Key / 模型由各宿主注入，**不会**画成节点端口，也不会写进流程图 JSON。 -默认打开可视化界面： +网页打开：（先启动 `MAF1.Web`） --- @@ -18,7 +18,8 @@ MAF1 是一个基于 **.NET 10** 和 **Microsoft Agents AI** 的多智能体（M - **进程外插件**：扫描输出目录 `plugins/*/plugin.json`，按 `launch` 启动子进程，stdin / stdout 走 JSON。 - **凭据隔离**：宿主配置 LLM；节点只声明需要哪种凭据（默认 `llm-default`）。浏览器拿不到原始 Key。 - **边条件**：连线可设 `hasValidCities` / `!hasValidCities`，不满足则跳过下游节点。 -- **CLI 对照**：`node` 模式把判断放在 Gate 节点内；`edge` 模式把判断写在 Workflow 边上。 +- **人机决策**：抽城市失败时暂停等待确认；必须有默认方案（结束查询）。网页弹出确认框，CLI 在控制台询问。超时未确认则自动采用默认方案，也可以改填其它城市继续查天气。 +- **CLI 对照**：`node` 模式把「有没有城市」写在 Gate 节点里；`edge` 模式把这条判断写在边上。两种模式缺城市时都会进入同一套确认。 - **天气数据源**：默认 [wttr.in](https://wttr.in)，可改为 OpenWeatherMap。 --- @@ -28,48 +29,43 @@ MAF1 是一个基于 **.NET 10** 和 **Microsoft Agents AI** 的多智能体（M | 部分 | 说明 | |------|------| | 运行时 | .NET 10（`net10.0`） | -| 宿主 | ASP.NET Core Web SDK，Minimal API + 静态文件 | -| 编排库 | `Microsoft.Agents.AI.Workflows` 1.19.0 | +| 控制台 | `MAF1`：普通控制台，Microsoft Agents AI Workflows | +| 网页 | `MAF1.Web`：ASP.NET Core Minimal API + `wwwroot` | | LLM | `Azure.AI.OpenAI` + `Microsoft.Agents.AI.OpenAI`（兼容 OpenAI / Azure OpenAI / DeepSeek 等） | -| 前端 | `wwwroot` 下原生 HTML / CSS / JS，无 npm 依赖 | -| 插件 | 独立控制台进程，协议见 `MAF1.Core/PluginContract` | +| 前端 | `MAF1.Web/wwwroot` 下原生 HTML / CSS / JS，无 npm 依赖 | +| 插件 | 独立控制台进程，协议见 `MAF1.Core/PluginContract`，由网页宿主扫描 | -解决方案文件：`MAF1.slnx`（包含宿主、`MAF1.Core`、两个插件工程）。 +解决方案文件：`MAF1.slnx`（`MAF1`、`MAF1.Web`、`MAF1.Core`、两个插件工程）。 --- ## 仓库结构 ``` -MAF1/ -├── MAF1.csproj # 宿主：Web UI + CLI +仓库根目录/ ├── MAF1.slnx -├── Program.cs # 有 CLI 参数则走命令行，否则启动 Web -├── CliHost.cs -├── appsettings.json # LLM / 插件 / 天气配置 -├── app.manifest # Windows 控制台 UTF-8 -├── Properties/launchSettings.json -├── Data/ # 示例文本 -│ ├── cities.txt -│ └── not-cities.txt -├── Web/ # 目录、图执行、运行时组装 -├── PluginHost/ # 扫描插件、起进程、注入凭据 -├── Workflows/ # CLI 用的 Microsoft Agents AI Workflow -├── Orchestration/ # CLI 事件流式输出 -├── wwwroot/ # 可视化编排界面 +├── MAF1/ # 控制台项目 +│ ├── MAF1.csproj +│ ├── Program.cs / CliHost.cs +│ ├── Workflows/ # Microsoft Agents AI Workflow +│ ├── Orchestration/ # 控制台事件流 + 确认提示 +│ ├── Data/ +│ └── appsettings.json +├── MAF1.Web/ # 网页编排项目 +│ ├── MAF1.Web.csproj +│ ├── Program.cs # Minimal API +│ ├── Web/ # 图执行、运行时组装 +│ ├── PluginHost/ # 扫描插件、起进程、注入凭据 +│ ├── wwwroot/ # 可视化界面 +│ ├── Data/ +│ └── appsettings.json ├── MAF1.Core/ # 共享：Agent、Tool、LLM、插件协议 -│ ├── Agents/FileCity/ -│ ├── Agents/Weather/ -│ ├── Tools/ -│ ├── Utils/AgentFactory.cs -│ └── PluginContract/ └── plugins/ - ├── README.md # 插件目录约定 - ├── file-city/ # FileCity 独立进程 - └── weather/ # Weather 独立进程 + ├── file-city/ + └── weather/ ``` -构建宿主时会编译两个插件，并把输出复制到宿主 `bin/.../plugins/file-city` 与 `plugins/weather`。**运行时扫描的是可执行文件旁边的 `plugins/`，不是源码树。** +构建 **MAF1.Web** 时会编译两个插件，并把输出复制到网页宿主 `bin/.../plugins/`。**运行时扫描的是网页 exe 旁边的 `plugins/`，不是源码树。** 控制台项目不扫描插件。 --- @@ -125,12 +121,14 @@ $env:OPENAI_CHAT_MODEL = "deepseek-v4-flash" Azure OpenAI：把 Endpoint 设成 Azure 资源地址，模型字段填 **部署名**。程序会按 URL 判断是否走 Azure 客户端。 -Visual Studio / `dotnet run --launch-profile designer` 会读 `Properties/launchSettings.json` 里的环境变量。该文件容易带上真实密钥，**请勿把含 Key 的版本推到远程**。 +Visual Studio / `dotnet run --project MAF1.Web --launch-profile designer` 会读 `MAF1.Web/Properties/launchSettings.json` 里的环境变量。该文件容易带上真实密钥，**请勿把含 Key 的版本推到远程**。 -### 3. 启动可视化编排（默认） +控制台 profile 在 `MAF1/Properties/launchSettings.json`（`node` / `edge`）。 + +### 3. 启动可视化编排 ```bash -dotnet run +dotnet run --project MAF1.Web ``` 浏览器打开 。控制台会打印：`可视化编排: http://127.0.0.1:5288`。 @@ -145,26 +143,43 @@ dotnet run ### 4. 命令行工作流 ```bash -dotnet run -- node Data/cities.txt # 条件在 Gate 节点内部 -dotnet run -- edge Data/cities.txt # 条件在边上分流 -dotnet run -- --help +dotnet run --project MAF1 -- node Data/cities.txt # 条件在 Gate 节点内部 +dotnet run --project MAF1 -- edge Data/cities.txt # 条件在边上分流 +dotnet run --project MAF1 -- --help ``` -无有效城市时可用 `Data/not-cities.txt` 看跳过天气查询的路径。 +无参数时默认 `node Data/cities.txt`。无有效城市时可用 `Data/not-cities.txt`。工作流会在控制台询问如何继续（默认结束查询，也可输入其它城市）；超时走默认方案。 -启动配置（`launchSettings.json`）： +```bash +dotnet run --project MAF1 -- node Data/not-cities.txt +dotnet run --project MAF1 -- edge Data/not-cities.txt +``` -| Profile | 作用 | -|---------|------| -| `designer` | Web 编排器（默认） | -| `node` | CLI，节点内判断 | -| `edge` | CLI，边上判断 | +控制台会出现： + +``` +[需要确认] +没有读到有效城市。… + 1) 结束查询（默认） ← 默认 + 2) 改查其它城市 +输入序号，或直接回车采用默认： +``` + +输入 `1` 或回车结束；输入 `2` 后再填城市名继续查天气；也可以直接输入 `成都`。等待秒数与网页相同，来自 `Workflow:DecisionTimeoutSeconds`。 + +启动配置： + +| 项目 | Profile | 作用 | +|------|---------|------| +| `MAF1.Web` | `designer` | 网页编排器 | +| `MAF1` | `node` | CLI，节点内判断 | +| `MAF1` | `edge` | CLI，边上判断 | --- ## 配置说明 -`appsettings.json` 会复制到输出目录。主要段落： +`MAF1/appsettings.json` 与 `MAF1.Web/appsettings.json` 会分别复制到各自输出目录。网页项目额外包含 `Plugins` / `Credentials`。主要段落： ### Llm @@ -192,6 +207,14 @@ dotnet run -- --help | `Wttr.UrlTemplate` | `{location}`、`{lang}` 占位 | | `OpenWeather.UrlTemplate` / `ApiKey` | OpenWeatherMap；需自行申请 Key | +### Workflow + +| 字段 | 含义 | +|------|------| +| `DecisionTimeoutSeconds` | 需要用户确认时的等待秒数，默认 `30`。超时后采用默认方案。 | + +节点 `config` 可覆盖：`decisionTimeoutSeconds`（秒）、`askOnEmptyCities`（`false` 时抽不到城市不询问，直接按边条件跳过）。 + --- ## 内置 Agent @@ -203,6 +226,7 @@ dotnet run -- --help - 输入：`filePath`（本地文本路径，相对宿主工作目录即可） - 输出：`hasValidCities`、`cities`、`reason`、`raw` - 行为：LLM + 读文件工具，从文本里抽出有效城市名 +- 若没有有效城市且后面还有节点：工作流会 **暂停等待确认**。默认方案是结束后续查询；可选方案是手动输入城市继续。超时走默认。 系统节点类型：`fileCity-system` 插件 id：`fileCity`（`plugins/file-city`） @@ -250,6 +274,7 @@ dotnet run -- --help ``` - **拓扑**：按依赖排序执行；未满足入边条件的节点会被标记为跳过。 +- **人机决策**：节点输出 `hasValidCities = false` 时，若后面还有节点，则返回 `status: needsDecision`，不立刻跑完。 - **端口**：`fromPort` → `toPort` 把上游输出接到下游输入；未接线的必填项可写在 `config`。 - **边条件 `when`**（当前实现）： - 空：始终通过 @@ -261,14 +286,39 @@ dotnet run -- --help ## HTTP API -端口固定：`http://127.0.0.1:5288`（`Program.cs`）。 +端口固定：`http://127.0.0.1:5288`（`MAF1.Web/Program.cs`）。 | 方法 | 路径 | 说明 | |------|------|------| | `GET` | `/api/catalog` | 系统节点 + 插件节点、扫描问题、`pluginsRoot` | | `GET` | `/api/credentials` | 对外凭据列表（不含原始 Key） | | `GET` | `/api/status` | 当前天气 Provider、插件根路径 | -| `POST` | `/api/run` | Body 为工作流图 JSON，返回逐步 `steps` | +| `POST` | `/api/run` | Body 为工作流图 JSON。可能直接跑完，也可能返回 `needsDecision` | +| `GET` | `/api/run/{runId}` | 查询暂停中或已结束的一次运行 | +| `POST` | `/api/run/{runId}/decide` | Body：`{ "optionId": "stop" }` 或 `{ "optionId": "query-cities", "text": "成都" }` | + +`POST /api/run` 在需要确认时返回： + +```json +{ + "ok": true, + "status": "needsDecision", + "runId": "…", + "decision": { + "prompt": "没有读到有效城市…", + "defaultOptionId": "stop", + "timeoutSeconds": 30, + "deadline": "2026-08-27T09:22:00+00:00", + "options": [ + { "id": "stop", "label": "结束查询（默认）", "isDefault": true }, + { "id": "query-cities", "label": "改查其它城市", "requiresText": true } + ] + }, + "steps": [] +} +``` + +默认方案必须存在。服务端到期会自动按 `defaultOptionId` 继续；客户端倒计时结束后可再 `GET` 一次拿最终结果。 JSON 使用 camelCase。静态站点来自 `wwwroot`。 @@ -336,7 +386,7 @@ JSON 使用 camelCase。静态站点来自 `wwwroot`。 ### 自己加插件 1. 新建 `plugins/你的插件/`，写 `plugin.json` 和启动程序。 -2. 若要随宿主一起编译，可仿照 `MAF1.csproj` 增加 `ProjectReference`（`ReferenceOutputAssembly=false`）和 `PublishPluginFolders` 复制规则。 +2. 若要随网页宿主一起编译，可仿照 `MAF1.Web.csproj` 增加 `ProjectReference`（`ReferenceOutputAssembly=false`）和 `PublishPluginFolders` 复制规则。 3. `id` 不要与系统节点类型冲突；若同名，目录会标记覆盖关系。 4. 重启宿主或点「刷新节点」，确认 `/api/catalog` 的 `issues` 为空。 @@ -345,10 +395,10 @@ JSON 使用 camelCase。静态站点来自 `wwwroot`。 ## 架构 ``` -浏览器 wwwroot +浏览器 MAF1.Web/wwwroot │ REST ▼ -ASP.NET Minimal API (Program.cs) +MAF1.Web Program.cs（Minimal API） │ ▼ AgentRuntime @@ -359,11 +409,11 @@ AgentRuntime └── PluginProcessRunner 子进程 stdin/stdout ``` -CLI 不走画布，直接： +控制台项目不走画布： -`CliHost` → `CityWeatherWorkflow`（node / edge）→ `WorkflowOrchestration` 把 Workflow 事件打到控制台。 +`MAF1` → `CliHost` → `CityWeatherWorkflow`（node / edge）→ `WorkflowOrchestration`。缺城市时走 `RequestPort`，与网页同一套默认方案 + 超时。 -`MAF1.Core` 被宿主与插件共用，避免两套 Agent 逻辑分叉。 +`MAF1.Core` 被控制台、网页与插件共用，避免两套 Agent 逻辑分叉。 --- @@ -377,9 +427,9 @@ CLI 不走画布，直接： 北京 ``` -`Data/not-cities.txt`：不含有效城市，用于验证「无城市则不查天气」。 +`Data/not-cities.txt`：不含有效城市，用于验证「无城市则询问后默认不查天气」。 -这些文件会随构建复制到输出目录。CLI 传入相对路径时，请在仓库根目录（或已复制 Data 的输出目录）下运行。 +这些文件会随构建复制到各项目输出目录。CLI 传入相对路径时，请在仓库根目录用 `--project MAF1` 运行，或在已复制 Data 的输出目录下运行。 --- @@ -389,7 +439,7 @@ CLI 不走画布，直接： `Llm:ApiKey` 和环境变量都为空。按「快速开始」配置后重启。 **画布上看不到插件** -插件必须出现在 **exe 旁边** 的 `plugins/`。先 `dotnet build`，确认 `bin/Debug/net10.0/plugins/file-city/plugin.json` 存在。源码目录里的 `plugins/` 不会被运行时直接扫描。 +插件必须出现在 **MAF1.Web.exe 旁边** 的 `plugins/`。先 `dotnet build MAF1.Web/MAF1.Web.csproj`，确认 `MAF1.Web/bin/Debug/net10.0/plugins/file-city/plugin.json` 存在。源码目录里的 `plugins/` 不会被运行时直接扫描。 **插件超时或卡住** 加大 `timeoutSeconds`；确认 LLM 与天气 HTTP 可访问；日志看 stderr。 @@ -401,7 +451,7 @@ CLI 不走画布，直接： 项目带 `app.manifest` 并在入口启用 UTF-8。若仍乱码，把终端代码页设为 UTF-8。 **端口被占用** -当前 URL 写死为 `127.0.0.1:5288`。关掉占用进程，或临时改 `Program.cs` / `launchSettings.json`。 +当前 URL 写死为 `127.0.0.1:5288`。关掉占用进程，或临时改 `MAF1.Web/Program.cs` / `MAF1.Web/Properties/launchSettings.json`。 **没有自动化测试 / Docker** 仓库目前没有测试项目和容器文件。验证方式：UI 示例图 + CLI `node` / `edge`。 diff --git a/Web/ConfigurableWorkflowRunner.cs b/Web/ConfigurableWorkflowRunner.cs deleted file mode 100644 index 48c0cf9..0000000 --- a/Web/ConfigurableWorkflowRunner.cs +++ /dev/null @@ -1,287 +0,0 @@ -using System.Text.Json; -using MAF1.Agents; -using MAF1.Agents.FileCity; -using MAF1.Agents.Weather; -using MAF1.PluginContract; -using MAF1.Plugins; -using Microsoft.Agents.AI; - -namespace MAF1.Web; - -public sealed class ConfigurableWorkflowRunner( - AIAgent fileCityAgent, - AIAgent weatherAgent, - NodeCatalogService catalog, - PluginProcessRunner plugins) -{ - private static readonly HashSet ReservedConfigKeys = new(StringComparer.OrdinalIgnoreCase) - { - "credentialId", - }; - - public async Task RunAsync(WorkflowGraph graph, CancellationToken cancellationToken) - { - try - { - Validate(graph); - NodeCatalogSnapshot snapshot = catalog.Load(); - List order = TopologicalOrder(graph); - Dictionary> outputs = []; - List steps = []; - - foreach (string nodeId in order) - { - WorkflowNode node = graph.Nodes.First(n => n.Id == nodeId); - Dictionary inputs = ResolveInputs(graph, node, outputs); - if (!PassEdgeConditions(graph, node, outputs, out string? skipReason)) - { - steps.Add(new NodeRunLog - { - NodeId = node.Id, - Type = node.Type, - Title = node.Title, - Skipped = true, - Message = skipReason, - Inputs = inputs, - }); - continue; - } - - NodeRunLog log = await RunNodeAsync(snapshot, node, inputs, cancellationToken); - steps.Add(log); - outputs[node.Id] = log.Outputs; - } - - return new WorkflowRunResult { Ok = true, Steps = steps }; - } - catch (Exception ex) - { - return new WorkflowRunResult { Ok = false, Error = ex.Message }; - } - } - - private async Task RunNodeAsync( - NodeCatalogSnapshot snapshot, - WorkflowNode node, - Dictionary inputs, - CancellationToken cancellationToken) - { - LoadedPlugin? plugin = snapshot.LoadedPlugins - .FirstOrDefault(p => p.Manifest.Id.Equals(node.Type, StringComparison.OrdinalIgnoreCase)); - if (plugin is not null) - { - return await RunPluginAsync(plugin, node, inputs, cancellationToken); - } - - AgentTypeInfo? system = AgentCatalog.FindSystem(node.Type) - ?? snapshot.System.FirstOrDefault(item => item.Type.Equals(node.Type, StringComparison.OrdinalIgnoreCase)); - if (system is not null) - { - AgentStepResult step = system.Handler switch - { - SystemHandler.FileCity => await FileCityAgent.RunAsync(fileCityAgent, inputs, cancellationToken), - SystemHandler.Weather => await WeatherAgent.RunAsync(weatherAgent, inputs, cancellationToken), - _ => throw new InvalidOperationException( - $"系统节点 `{system.Type}` 在 AgentCatalog 里没有绑定实现。请给它设置 Handler。"), - }; - return ToLog(node, step); - } - - throw new InvalidOperationException( - $"未找到节点类型 `{node.Type}`。系统节点来自 AgentCatalog，插件节点来自 plugins 目录，请点「刷新节点」。"); - } - - private async Task RunPluginAsync( - LoadedPlugin plugin, - WorkflowNode node, - Dictionary inputs, - CancellationToken cancellationToken) - { - Dictionary declared = FilterDeclaredInputs(plugin.Manifest, inputs); - node.Config.TryGetValue("credentialId", out string? credentialId); - Dictionary outputs = await plugins.RunAsync(plugin, declared, credentialId, cancellationToken); - string? stderr = null; - if (outputs.Remove("_stderr", out object? stderrValue)) - { - stderr = stderrValue?.ToString(); - } - - return new NodeRunLog - { - NodeId = node.Id, - Type = node.Type, - Title = node.Title, - Message = stderr, - Inputs = declared, - Outputs = outputs, - }; - } - - private static NodeRunLog ToLog(WorkflowNode node, AgentStepResult step) - => new() - { - NodeId = node.Id, - Type = node.Type, - Title = node.Title, - Message = step.Message, - Inputs = step.Inputs, - Outputs = step.Outputs, - }; - - private static Dictionary FilterDeclaredInputs(PluginManifest manifest, Dictionary inputs) - { - Dictionary declared = new(StringComparer.OrdinalIgnoreCase); - foreach (PluginPort port in manifest.Inputs) - { - if (inputs.TryGetValue(port.Name, out object? value)) - { - declared[port.Name] = value; - } - } - - return declared; - } - - private static void Validate(WorkflowGraph graph) - { - if (graph.Nodes.Count == 0) - { - throw new InvalidOperationException("画布上还没有节点。"); - } - - HashSet ids = graph.Nodes.Select(n => n.Id).ToHashSet(); - foreach (WorkflowEdge edge in graph.Edges) - { - if (!ids.Contains(edge.From) || !ids.Contains(edge.To)) - { - throw new InvalidOperationException("存在指向已删除节点的连线。"); - } - } - } - - private static List TopologicalOrder(WorkflowGraph graph) - { - Dictionary indegree = graph.Nodes.ToDictionary(n => n.Id, _ => 0); - Dictionary> outgoing = graph.Nodes.ToDictionary(n => n.Id, _ => new List()); - foreach (WorkflowEdge edge in graph.Edges) - { - indegree[edge.To]++; - outgoing[edge.From].Add(edge.To); - } - - Queue ready = new(indegree.Where(kv => kv.Value == 0).Select(kv => kv.Key)); - List order = []; - while (ready.Count > 0) - { - string id = ready.Dequeue(); - order.Add(id); - foreach (string next in outgoing[id].Distinct()) - { - indegree[next]--; - if (indegree[next] == 0) - { - ready.Enqueue(next); - } - } - } - - if (order.Count != graph.Nodes.Count) - { - throw new InvalidOperationException("工作流存在环，无法运行。"); - } - - return order; - } - - private static Dictionary ResolveInputs( - WorkflowGraph graph, - WorkflowNode node, - Dictionary> outputs) - { - Dictionary inputs = new(StringComparer.OrdinalIgnoreCase); - foreach (KeyValuePair item in node.Config) - { - if (ReservedConfigKeys.Contains(item.Key) || string.IsNullOrWhiteSpace(item.Value)) - { - continue; - } - - inputs[item.Key] = item.Value; - } - - foreach (WorkflowEdge edge in graph.Edges.Where(e => e.To == node.Id)) - { - if (!outputs.TryGetValue(edge.From, out Dictionary? source)) - { - continue; - } - - if (source.TryGetValue(edge.FromPort, out object? value)) - { - inputs[edge.ToPort] = value; - } - } - - return inputs; - } - - private static bool PassEdgeConditions( - WorkflowGraph graph, - WorkflowNode node, - Dictionary> outputs, - out string? skipReason) - { - List incoming = graph.Edges.Where(e => e.To == node.Id).ToList(); - if (incoming.Count == 0) - { - skipReason = null; - return true; - } - - foreach (WorkflowEdge edge in incoming) - { - if (string.IsNullOrWhiteSpace(edge.When)) - { - continue; - } - - if (!outputs.TryGetValue(edge.From, out Dictionary? source)) - { - skipReason = $"上一节点 {edge.From} 尚未产出结果。"; - return false; - } - - bool hasCities = ReadBool(source, "hasValidCities"); - bool pass = edge.When switch - { - "hasValidCities" => hasCities, - "!hasValidCities" => !hasCities, - _ => true, - }; - if (!pass) - { - skipReason = $"连线条件 {edge.When} 不满足，跳过本节点。"; - return false; - } - } - - skipReason = null; - return true; - } - - private static bool ReadBool(Dictionary map, string key) - { - if (!map.TryGetValue(key, out object? value) || value is null) - { - return false; - } - - return value switch - { - bool b => b, - JsonElement el when el.ValueKind == JsonValueKind.True => true, - JsonElement el when el.ValueKind == JsonValueKind.False => false, - _ => bool.TryParse(value.ToString(), out bool parsed) && parsed, - }; - } -} diff --git a/Workflows/SkipWeatherExecutor.cs b/Workflows/SkipWeatherExecutor.cs deleted file mode 100644 index 7193e25..0000000 --- a/Workflows/SkipWeatherExecutor.cs +++ /dev/null @@ -1,28 +0,0 @@ -using MAF1.Agents.FileCity; -using Microsoft.Agents.AI.Workflows; - -namespace MAF1.Workflows; - -/// -/// 只有「没有城市」那条边会进到这里，所以这里不再判断。 -/// -internal sealed class SkipWeatherExecutor() : Executor("SkipWeather") -{ - protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder builder) - { - return base.ConfigureProtocol(builder).YieldsOutput(); - } - - public override async ValueTask HandleAsync( - CityExtraction extraction, - IWorkflowContext context, - CancellationToken cancellationToken = default) - { - string reason = string.IsNullOrWhiteSpace(extraction.Reason) - ? "文件里没有可查询的城市。" - : extraction.Reason; - await context.YieldOutputAsync( - $"没有有效城市，工作流结束，不执行 WeatherAgent。{reason}", - cancellationToken); - } -}