288 lines
9.4 KiB
C#
288 lines
9.4 KiB
C#
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<string> ReservedConfigKeys = new(StringComparer.OrdinalIgnoreCase)
|
|
{
|
|
"credentialId",
|
|
};
|
|
|
|
public async Task<WorkflowRunResult> RunAsync(WorkflowGraph graph, CancellationToken cancellationToken)
|
|
{
|
|
try
|
|
{
|
|
Validate(graph);
|
|
NodeCatalogSnapshot snapshot = catalog.Load();
|
|
List<string> order = TopologicalOrder(graph);
|
|
Dictionary<string, Dictionary<string, object?>> outputs = [];
|
|
List<NodeRunLog> steps = [];
|
|
|
|
foreach (string nodeId in order)
|
|
{
|
|
WorkflowNode node = graph.Nodes.First(n => n.Id == nodeId);
|
|
Dictionary<string, object?> 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<NodeRunLog> RunNodeAsync(
|
|
NodeCatalogSnapshot snapshot,
|
|
WorkflowNode node,
|
|
Dictionary<string, object?> 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<NodeRunLog> RunPluginAsync(
|
|
LoadedPlugin plugin,
|
|
WorkflowNode node,
|
|
Dictionary<string, object?> inputs,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
Dictionary<string, object?> declared = FilterDeclaredInputs(plugin.Manifest, inputs);
|
|
node.Config.TryGetValue("credentialId", out string? credentialId);
|
|
Dictionary<string, object?> 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<string, object?> FilterDeclaredInputs(PluginManifest manifest, Dictionary<string, object?> inputs)
|
|
{
|
|
Dictionary<string, object?> 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<string> 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<string> TopologicalOrder(WorkflowGraph graph)
|
|
{
|
|
Dictionary<string, int> indegree = graph.Nodes.ToDictionary(n => n.Id, _ => 0);
|
|
Dictionary<string, List<string>> outgoing = graph.Nodes.ToDictionary(n => n.Id, _ => new List<string>());
|
|
foreach (WorkflowEdge edge in graph.Edges)
|
|
{
|
|
indegree[edge.To]++;
|
|
outgoing[edge.From].Add(edge.To);
|
|
}
|
|
|
|
Queue<string> ready = new(indegree.Where(kv => kv.Value == 0).Select(kv => kv.Key));
|
|
List<string> 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<string, object?> ResolveInputs(
|
|
WorkflowGraph graph,
|
|
WorkflowNode node,
|
|
Dictionary<string, Dictionary<string, object?>> outputs)
|
|
{
|
|
Dictionary<string, object?> inputs = new(StringComparer.OrdinalIgnoreCase);
|
|
foreach (KeyValuePair<string, string> 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<string, object?>? 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<string, Dictionary<string, object?>> outputs,
|
|
out string? skipReason)
|
|
{
|
|
List<WorkflowEdge> 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<string, object?>? 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<string, object?> 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,
|
|
};
|
|
}
|
|
}
|