mirror of
https://github.com/open-jarvis/OpenJarvis.git
synced 2026-07-31 03:12:16 +00:00
321 lines
11 KiB
Python
321 lines
11 KiB
Python
"""WorkflowEngine — executes a WorkflowGraph against a JarvisSystem."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import concurrent.futures
|
|
import time
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from openjarvis.core.events import EventBus, EventType
|
|
from openjarvis.workflow.graph import WorkflowGraph
|
|
from openjarvis.workflow.types import (
|
|
NodeType,
|
|
WorkflowNode,
|
|
WorkflowResult,
|
|
WorkflowStepResult,
|
|
)
|
|
|
|
|
|
class WorkflowEngine:
|
|
"""Execute DAG-based workflows.
|
|
|
|
Sequential nodes run in topological order. Parallel-eligible nodes
|
|
(same execution stage, no inter-dependencies) run via ThreadPoolExecutor.
|
|
Condition nodes evaluate expressions against prior step outputs.
|
|
Loop nodes use LoopGuard from Phase 14.3.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
bus: Optional[EventBus] = None,
|
|
max_parallel: int = 4,
|
|
default_node_timeout: int = 300,
|
|
) -> None:
|
|
self._bus = bus
|
|
self._max_parallel = max_parallel
|
|
self._default_node_timeout = default_node_timeout
|
|
|
|
def run(
|
|
self,
|
|
graph: WorkflowGraph,
|
|
system: Any = None, # JarvisSystem
|
|
*,
|
|
initial_input: str = "",
|
|
context: Optional[Dict[str, Any]] = None,
|
|
) -> WorkflowResult:
|
|
"""Execute a workflow graph end-to-end."""
|
|
valid, msg = graph.validate()
|
|
if not valid:
|
|
return WorkflowResult(
|
|
workflow_name=graph.name,
|
|
success=False,
|
|
final_output=f"Invalid workflow: {msg}",
|
|
)
|
|
|
|
t0 = time.time()
|
|
if self._bus:
|
|
self._bus.publish(
|
|
EventType.WORKFLOW_START,
|
|
{"workflow": graph.name},
|
|
)
|
|
|
|
# State: outputs keyed by node_id
|
|
outputs: Dict[str, str] = {"_input": initial_input}
|
|
ctx = dict(context or {})
|
|
all_steps: List[WorkflowStepResult] = []
|
|
success = True
|
|
|
|
stages = graph.execution_stages()
|
|
for stage in stages:
|
|
if len(stage) == 1:
|
|
# Sequential execution
|
|
step = self._execute_node(
|
|
graph.get_node(stage[0]), # type: ignore[arg-type]
|
|
outputs,
|
|
ctx,
|
|
system,
|
|
graph,
|
|
)
|
|
all_steps.append(step)
|
|
outputs[stage[0]] = step.output
|
|
if not step.success:
|
|
success = False
|
|
break
|
|
else:
|
|
# Parallel execution
|
|
with concurrent.futures.ThreadPoolExecutor(
|
|
max_workers=min(len(stage), self._max_parallel),
|
|
) as pool:
|
|
futures = {
|
|
pool.submit(
|
|
self._execute_node,
|
|
graph.get_node(nid),
|
|
dict(outputs),
|
|
dict(ctx),
|
|
system,
|
|
graph,
|
|
): nid
|
|
for nid in stage
|
|
}
|
|
for future in concurrent.futures.as_completed(futures):
|
|
nid = futures[future]
|
|
try:
|
|
step = future.result(
|
|
timeout=self._default_node_timeout,
|
|
)
|
|
except Exception as exc:
|
|
step = WorkflowStepResult(
|
|
node_id=nid,
|
|
success=False,
|
|
output=f"Node execution error: {exc}",
|
|
)
|
|
all_steps.append(step)
|
|
outputs[nid] = step.output
|
|
if not step.success:
|
|
success = False
|
|
|
|
if not success:
|
|
break
|
|
|
|
total = time.time() - t0
|
|
# Final output is the output of the last executed node
|
|
final_output = all_steps[-1].output if all_steps else ""
|
|
|
|
if self._bus:
|
|
self._bus.publish(
|
|
EventType.WORKFLOW_END,
|
|
{"workflow": graph.name, "success": success, "duration": total},
|
|
)
|
|
|
|
return WorkflowResult(
|
|
workflow_name=graph.name,
|
|
success=success,
|
|
steps=all_steps,
|
|
final_output=final_output,
|
|
total_duration_seconds=total,
|
|
)
|
|
|
|
def _execute_node(
|
|
self,
|
|
node: WorkflowNode,
|
|
outputs: Dict[str, str],
|
|
ctx: Dict[str, Any],
|
|
system: Any,
|
|
graph: WorkflowGraph,
|
|
) -> WorkflowStepResult:
|
|
"""Execute a single workflow node."""
|
|
if self._bus:
|
|
self._bus.publish(
|
|
EventType.WORKFLOW_NODE_START,
|
|
{"node": node.id, "type": node.node_type.value},
|
|
)
|
|
|
|
t0 = time.time()
|
|
try:
|
|
if node.node_type == NodeType.AGENT:
|
|
result = self._run_agent_node(node, outputs, system, graph)
|
|
elif node.node_type == NodeType.TOOL:
|
|
result = self._run_tool_node(node, outputs, system)
|
|
elif node.node_type == NodeType.CONDITION:
|
|
result = self._run_condition_node(node, outputs)
|
|
elif node.node_type == NodeType.TRANSFORM:
|
|
result = self._run_transform_node(node, outputs)
|
|
elif node.node_type == NodeType.LOOP:
|
|
result = self._run_loop_node(node, outputs, system, graph)
|
|
else:
|
|
result = WorkflowStepResult(
|
|
node_id=node.id,
|
|
success=False,
|
|
output=f"Unknown node type: {node.node_type}",
|
|
)
|
|
except Exception as exc:
|
|
result = WorkflowStepResult(
|
|
node_id=node.id,
|
|
success=False,
|
|
output=f"Node error: {exc}",
|
|
)
|
|
|
|
result.duration_seconds = time.time() - t0
|
|
|
|
if self._bus:
|
|
self._bus.publish(
|
|
EventType.WORKFLOW_NODE_END,
|
|
{
|
|
"node": node.id,
|
|
"success": result.success,
|
|
"duration": result.duration_seconds,
|
|
},
|
|
)
|
|
|
|
return result
|
|
|
|
def _get_node_input(
|
|
self, node: WorkflowNode, outputs: Dict[str, str], graph: WorkflowGraph,
|
|
) -> str:
|
|
"""Get input for a node from predecessor outputs."""
|
|
preds = graph.predecessors(node.id)
|
|
if preds:
|
|
parts = [outputs.get(p, "") for p in preds if outputs.get(p)]
|
|
return "\n\n".join(parts) if parts else outputs.get("_input", "")
|
|
return outputs.get("_input", "")
|
|
|
|
def _run_agent_node(
|
|
self, node: WorkflowNode, outputs: Dict[str, str],
|
|
system: Any, graph: WorkflowGraph,
|
|
) -> WorkflowStepResult:
|
|
"""Execute an agent node."""
|
|
input_text = self._get_node_input(node, outputs, graph)
|
|
if system is None:
|
|
return WorkflowStepResult(
|
|
node_id=node.id,
|
|
success=False,
|
|
output="No system available for agent execution.",
|
|
)
|
|
try:
|
|
result = system.ask(
|
|
input_text,
|
|
agent=node.agent or None,
|
|
tools=node.tools or None,
|
|
)
|
|
return WorkflowStepResult(
|
|
node_id=node.id,
|
|
success=True,
|
|
output=result.get("content", ""),
|
|
)
|
|
except Exception as exc:
|
|
return WorkflowStepResult(
|
|
node_id=node.id,
|
|
success=False,
|
|
output=f"Agent error: {exc}",
|
|
)
|
|
|
|
def _run_tool_node(
|
|
self, node: WorkflowNode, outputs: Dict[str, str], system: Any,
|
|
) -> WorkflowStepResult:
|
|
"""Execute a tool node."""
|
|
tool_name = node.config.get("tool_name", "")
|
|
tool_args = node.config.get("tool_args", "{}")
|
|
if system and system.tool_executor:
|
|
from openjarvis.core.types import ToolCall
|
|
tc = ToolCall(id=f"wf_{node.id}", name=tool_name, arguments=tool_args)
|
|
tr = system.tool_executor.execute(tc)
|
|
return WorkflowStepResult(
|
|
node_id=node.id,
|
|
success=tr.success,
|
|
output=tr.content,
|
|
)
|
|
return WorkflowStepResult(
|
|
node_id=node.id,
|
|
success=False,
|
|
output="No tool executor available.",
|
|
)
|
|
|
|
def _run_condition_node(
|
|
self, node: WorkflowNode, outputs: Dict[str, str],
|
|
) -> WorkflowStepResult:
|
|
"""Evaluate a condition expression against outputs."""
|
|
expr = node.condition_expr
|
|
if not expr:
|
|
return WorkflowStepResult(
|
|
node_id=node.id, success=True, output="true",
|
|
)
|
|
# Simple expression evaluation — check if key exists and is truthy
|
|
# Supports: "node_id.success", "node_id.output contains 'text'"
|
|
try:
|
|
result = str(eval(expr, {"__builtins__": {}}, {"outputs": outputs})) # noqa: S307
|
|
except Exception:
|
|
result = "false"
|
|
return WorkflowStepResult(
|
|
node_id=node.id,
|
|
success=True,
|
|
output=result,
|
|
)
|
|
|
|
def _run_transform_node(
|
|
self, node: WorkflowNode, outputs: Dict[str, str],
|
|
) -> WorkflowStepResult:
|
|
"""Apply a text transformation."""
|
|
expr = node.transform_expr
|
|
preds = [outputs.get(p, "") for p in outputs if p != "_input"]
|
|
combined = "\n\n".join(preds) if preds else ""
|
|
if expr == "concatenate":
|
|
return WorkflowStepResult(node_id=node.id, output=combined)
|
|
if expr == "first_line":
|
|
return WorkflowStepResult(
|
|
node_id=node.id,
|
|
output=combined.split("\n")[0] if combined else "",
|
|
)
|
|
return WorkflowStepResult(node_id=node.id, output=combined)
|
|
|
|
def _run_loop_node(
|
|
self, node: WorkflowNode, outputs: Dict[str, str],
|
|
system: Any, graph: WorkflowGraph,
|
|
) -> WorkflowStepResult:
|
|
"""Execute a loop node (re-runs agent until condition or max iterations)."""
|
|
input_text = self._get_node_input(node, outputs, graph)
|
|
max_iter = node.max_iterations
|
|
last_output = input_text
|
|
for i in range(max_iter):
|
|
if system:
|
|
result = system.ask(last_output, agent=node.agent or None)
|
|
last_output = result.get("content", "")
|
|
# Check if loop should terminate
|
|
if (
|
|
node.condition_expr
|
|
and node.condition_expr.lower()
|
|
in last_output.lower()
|
|
):
|
|
break
|
|
else:
|
|
break
|
|
return WorkflowStepResult(
|
|
node_id=node.id,
|
|
success=True,
|
|
output=last_output,
|
|
metadata={"iterations": i + 1},
|
|
)
|
|
|
|
|
|
__all__ = ["WorkflowEngine"]
|