lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

foreach!

Browse Source 
using the new frame primitive. it spawns n of them on the underlying container ForeachNode.over, spawns a new frame with fresh ahh context each loop iteration, and the printed result looks so goo yo
This commit is contained in:
lda
2026-04-28 13:10:52 +07:00 Verified
parent 9f24c61889
commit 04225874ea
5 changed files with 229 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+57 -12
View File
@@ -26,6 +26,7 @@ workflow = Workflow.model_validate(
"folder_id": {"type": "string"},
"should_email": {"type": "boolean"},
"documents": {"type": "array", "merge_strategy": "replace"},
"item_summaries": {"type": "array", "merge_strategy": "append"},
"summary": {"type": "string", "merge_strategy": "replace"},
"approved": {"type": "boolean", "merge_strategy": "replace"},
"approval_comment": {"type": "string", "merge_strategy": "replace"},
@@ -56,11 +57,25 @@ workflow = Workflow.model_validate(
"outcomes": ["ok"],
},
{
"name": "summarize_documents",
"name": "summarize_document",
"input_schema": {
"type": "object",
"properties": {"documents": {"type": "array"}},
"required": ["documents"],
"properties": {"document": {"type": "string"}},
"required": ["document"],
},
"output_schema": {
"type": "object",
"properties": {"item_summary": {"type": "string"}},
"required": ["item_summary"],
},
"outcomes": ["ok"],
},
{
"name": "combine_summaries",
"input_schema": {
"type": "object",
"properties": {"item_summaries": {"type": "array"}},
"required": ["item_summaries"],
},
"output_schema": {
"type": "object",
@@ -108,11 +123,27 @@ workflow = Workflow.model_validate(
"out_map": {"documents": "state.documents"},
},
{
"id": "summarize",
"id": "summarize_each",
"type": "foreach",
"over": "state.documents",
"as": "document",
"mode": "serial",
"on_item_error": "fail",
},
{
"id": "summarize_one",
"type": "node",
"node": "summarize_documents",
"desc": "Summarize all retrieved documents",
"in_map": {"state.documents": "documents"},
"node": "summarize_document",
"desc": "Summarize one document",
"in_map": {"context.document": "document"},
"out_map": {"item_summary": "state.item_summaries"},
},
{
"id": "combine_summaries",
"type": "node",
"node": "combine_summaries",
"desc": "Combine item summaries into one final summary",
"in_map": {"state.item_summaries": "item_summaries"},
"out_map": {"summary": "state.summary"},
},
{
@@ -155,8 +186,11 @@ workflow = Workflow.model_validate(
},
],
"edges": [
{"from": "list_files", "outcome": "ok", "to": "summarize"},
{"from": "summarize", "outcome": "ok", "to": "should_email"},
{"from": "list_files", "outcome": "ok", "to": "summarize_each"},
{"from": "summarize_each", "outcome": "loop", "to": "summarize_one"},
{"from": "summarize_each", "outcome": "done", "to": "combine_summaries"},
{"from": "summarize_one", "outcome": "ok", "to": END},
{"from": "combine_summaries", "outcome": "ok", "to": "should_email"},
{"from": "should_email", "outcome": "true", "to": "approve_email"},
{"from": "should_email", "outcome": "false", "to": "skip_email"},
{"from": "approve_email", "outcome": "submitted", "to": "send_email"},
@@ -191,10 +225,20 @@ def drive_list_files(
def summarize_documents(
payload: dict[str, object], ctx: RuntimeContext
) -> dict[str, object]:
documents = payload["documents"]
document = payload["document"]
return {
"outcome": "ok",
"output": {"summary": f"Summarized {len(documents)} documents from Drive."},
"output": {"item_summary": f"Summary of {document}"},
}
def combine_summaries(
payload: dict[str, object], ctx: RuntimeContext
) -> dict[str, object]:
item_summaries = payload["item_summaries"]
return {
"outcome": "ok",
"output": {"summary": " | ".join(item_summaries)},
}
@@ -216,7 +260,8 @@ def mark_email_skipped(
registry = {
"drive_list_files": drive_list_files,
"summarize_documents": summarize_documents,
"summarize_document": summarize_documents,
"combine_summaries": combine_summaries,
"send_email": send_email,
"mark_email_skipped": mark_email_skipped,
}
wf_core/__init__.py
+2
View File
@@ -21,6 +21,7 @@ from .runtime import (
)
from .run_state import (
ExecutionFrame,
FrameStatus,
InterruptRequest,
RunState,
RunStatus,
@@ -48,6 +49,7 @@ __all__ = [
"StateSchema",
"NodeHandler",
"ExecutionFrame",
"FrameStatus",
"RunState",
"RunStatus",
"RuntimeContext",
wf_core/run_state.py
+11
View File
@@ -13,15 +13,25 @@ class RunStatus(StrEnum):
INTERRUPTED = "interrupted"
class FrameStatus(StrEnum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
INTERRUPTED = "interrupted"
@dataclass(slots=True)
class ExecutionFrame:
id: str
kind: str
node_id: str
status: FrameStatus = FrameStatus.PENDING
parent_frame_id: str | None = None
prior_outcome: str | None = None
activated_incoming_edge: str | None = None
metadata: dict[str, Any] = field(default_factory=dict)
finished_at_node_id: str | None = None
@dataclass(slots=True)
@@ -31,6 +41,7 @@ class RuntimeContext:
retry_count: int = 0
prior_outcome: str | None = None
activated_incoming_edge: str | None = None
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass(slots=True)
wf_core/runtime.py
+154 -8
View File
@@ -17,6 +17,7 @@ from .model import (
)
from .run_state import (
ExecutionFrame,
FrameStatus,
InterruptRequest,
RunState,
RunStatus,
@@ -46,6 +47,7 @@ def execute_workflow(
id="root",
kind="workflow",
node_id=workflow.start,
status=FrameStatus.PENDING,
)
},
current_frame_id="root",
@@ -80,7 +82,7 @@ def resume_workflow(
if run.current_frame_id is None:
raise WorkflowExecutionError("run has no current frame")
run.sync_from_current_frame()
_collapse_completed_frames(run)
if run.current_node_id is None:
raise WorkflowExecutionError("run has no current node")
@@ -103,6 +105,7 @@ def resume_workflow(
resume_payload=resume_payload,
resume_outcome=resume_outcome,
)
_collapse_completed_frames(run)
if run.current_node_id == END:
run.output = project_output(workflow, run.state)
validate_payload_against_schema(
@@ -113,8 +116,12 @@ def resume_workflow(
run.status = RunStatus.RUNNING
run.error = None
run.current_frame().status = FrameStatus.RUNNING
while run.current_node_id != END:
while True:
_collapse_completed_frames(run)
if run.current_node_id == END:
break
step_workflow(
workflow,
run,
@@ -147,7 +154,7 @@ def step_workflow(
if run.current_frame_id is None:
raise WorkflowExecutionError("run has no current frame")
run.sync_from_current_frame()
_collapse_completed_frames(run)
if run.current_node_id is None or run.current_node_id == END:
return run
if run.status == RunStatus.INTERRUPTED:
@@ -166,6 +173,8 @@ def step_workflow(
}
frame = run.current_frame()
if frame.status == FrameStatus.PENDING:
frame.status = FrameStatus.RUNNING
step = nodes_by_id[frame.node_id]
if isinstance(step, NodeUse):
@@ -174,7 +183,10 @@ def step_workflow(
outcome = step_result["outcome"]
elif isinstance(step, ConditionNode):
predicate = eval_condition(
step.check, run.state, run.workflow_input, frame.prior_outcome
step.check,
run.state,
run.workflow_input,
frame.prior_outcome,
)
outcome = "true" if predicate else "false"
step_result = {
@@ -195,9 +207,11 @@ def step_workflow(
frame_id=frame.id,
state=run.state,
workflow_input=run.workflow_input,
context=_frame_context_values(frame),
)
run.interrupt = interrupt_request
run.status = RunStatus.INTERRUPTED
frame.status = FrameStatus.INTERRUPTED
run.trace.append(
TraceEntry(
frame_id=frame.id,
@@ -212,7 +226,7 @@ def step_workflow(
)
return run
elif isinstance(step, ForeachNode):
raise WorkflowExecutionError("foreach execution is not implemented yet")
return _step_foreach(workflow, run, step, edge_map)
else:
raise WorkflowExecutionError(f"unsupported step type {step.type!r}")
@@ -238,6 +252,101 @@ def step_workflow(
frame.prior_outcome = outcome
frame.activated_incoming_edge = frame.node_id
frame.node_id = next_node_id
if next_node_id == END:
frame.status = FrameStatus.COMPLETED
frame.finished_at_node_id = END
run.sync_from_current_frame()
return run
def _step_foreach(
workflow: Workflow,
run: RunState,
step: ForeachNode,
edge_map: dict[tuple[str, str], str],
) -> RunState:
if step.mode != "serial":
raise WorkflowExecutionError(
"parallel foreach execution is not implemented yet"
)
frame = run.current_frame()
progress_map = frame.metadata.setdefault("foreach_progress", {})
progress = progress_map.setdefault(step.id, {"index": 0})
iterable = safe_resolve_path(
step.over,
state=run.state,
workflow_input=run.workflow_input,
context=_frame_context_values(frame),
)
if not isinstance(iterable, list):
raise WorkflowExecutionError(
f"foreach source {step.over!r} must resolve to a list"
)
index = progress["index"]
if index >= len(iterable):
outcome = "done"
next_node_id = edge_map.get((frame.node_id, outcome))
if next_node_id is None:
raise WorkflowExecutionError(
f"no edge found for node {frame.node_id!r} and outcome {outcome!r}"
)
run.trace.append(
TraceEntry(
frame_id=frame.id,
node_id=frame.node_id,
step_type=step.type,
resolved_input={"count": len(iterable), "index": index},
outcome=outcome,
next_node_id=next_node_id,
output={},
state_changes={},
)
)
frame.prior_outcome = outcome
frame.activated_incoming_edge = frame.node_id
frame.node_id = next_node_id
run.sync_from_current_frame()
return run
loop_start = edge_map.get((frame.node_id, "loop"))
if loop_start is None:
raise WorkflowExecutionError(
f"no edge found for foreach node {frame.node_id!r} and outcome 'loop'"
)
item = iterable[index]
progress["index"] = index + 1
child_id = f"{frame.id}:{step.id}:{index}"
child_metadata = {
"foreach_node_id": step.id,
"loop_index": index,
"loop_item": item,
"loop_alias": step.as_,
}
run.frames[child_id] = ExecutionFrame(
id=child_id,
kind="foreach_iteration",
node_id=loop_start,
status=FrameStatus.PENDING,
parent_frame_id=frame.id,
metadata=child_metadata,
)
run.trace.append(
TraceEntry(
frame_id=frame.id,
node_id=frame.node_id,
step_type=step.type,
resolved_input={"item": item, "index": index},
outcome="loop",
next_node_id=loop_start,
output={},
state_changes={},
)
)
run.current_frame_id = child_id
run.sync_from_current_frame()
return run
@@ -255,12 +364,14 @@ def _execute_node_use(
f"no handler registered for node def {node.node!r}"
)
frame = run.current_frame()
context_values = _frame_context_values(frame)
resolved_input = {
destination_field: safe_resolve_path(
source_path,
state=run.state,
workflow_input=run.workflow_input,
context={},
context=context_values,
)
for source_path, destination_field in node.in_map.items()
}
@@ -268,12 +379,12 @@ def _execute_node_use(
node_def.input_schema, resolved_input, f"node input for {node.id}"
)
frame = run.current_frame()
context = RuntimeContext(
current_node_id=node.id,
frame_id=frame.id,
prior_outcome=frame.prior_outcome,
activated_incoming_edge=frame.activated_incoming_edge,
metadata=dict(frame.metadata),
)
raw_result = handler(resolved_input, context)
result = coerce_node_result(raw_result)
@@ -309,13 +420,14 @@ def _build_interrupt_request(
frame_id: str,
state: dict[str, Any],
workflow_input: dict[str, Any],
context: dict[str, Any],
) -> InterruptRequest:
payload = {
payload_field: safe_resolve_path(
source_path,
state=state,
workflow_input=workflow_input,
context={},
context=context,
)
for source_path, payload_field in node.request_map.items()
}
@@ -383,5 +495,39 @@ def _resume_interrupt(
frame.prior_outcome = resume_outcome
frame.activated_incoming_edge = frame.node_id
frame.node_id = next_node_id
frame.status = FrameStatus.RUNNING if next_node_id != END else FrameStatus.COMPLETED
frame.finished_at_node_id = END if next_node_id == END else None
run.interrupt = None
run.sync_from_current_frame()
def _collapse_completed_frames(run: RunState) -> None:
while run.current_frame_id is not None:
frame = run.current_frame()
if frame.node_id == END and frame.status != FrameStatus.COMPLETED:
frame.status = FrameStatus.COMPLETED
frame.finished_at_node_id = END
if frame.status != FrameStatus.COMPLETED or frame.parent_frame_id is None:
run.sync_from_current_frame()
return
run.current_frame_id = frame.parent_frame_id
parent = run.current_frame()
if parent.status == FrameStatus.PENDING:
parent.status = FrameStatus.RUNNING
run.sync_from_current_frame()
def _frame_context_values(frame: ExecutionFrame) -> dict[str, Any]:
context: dict[str, Any] = {
"prior_outcome": frame.prior_outcome,
"activated_incoming_edge": frame.activated_incoming_edge,
}
if frame.kind == "foreach_iteration":
loop_item = frame.metadata.get("loop_item")
loop_index = frame.metadata.get("loop_index")
loop_alias = frame.metadata.get("loop_alias")
context["loop_item"] = loop_item
context["loop_index"] = loop_index
if isinstance(loop_alias, str) and loop_alias:
context[loop_alias] = loop_item
return context
wf_core/validate.py
+5 -3
View File
@@ -205,11 +205,13 @@ def _validate_node_use(
f"nodes[{index}].in_map[{source_path!r}]",
f"destination field {destination_field!r} is not declared in node input schema",
)
if not is_valid_source_path(source_path, state_fields, input_root_fields):
if not is_valid_source_path(
source_path, state_fields, input_root_fields, allow_context=True
):
report.add(
ValidationIssueCode.INVALID_SOURCE_PATH,
f"nodes[{index}].in_map[{source_path!r}]",
"source path must start with input. or state. and reference a declared root field",
"source path must start with input., state., or context. and reference a declared root field when applicable",
)
for source_field, destination_path in node.out_map.items():
@@ -385,7 +387,7 @@ def _declared_outcomes_for_step(step: Step, node_defs: dict[str, NodeDef]) -> se
if step.type == "condition":
return {"true", "false"}
if step.type == "foreach":
return {"done"}
return {"loop", "done"}
if step.type == "join":
return {"done"}
if step.type == "interrupt":