Agents: Build a Deterministic Workflow#

This advanced notebook constructs the orchestration that the simple pattern tutorial deliberately avoided. Two typed LogicStep handlers form a DAG: one scales scores and the dependent step summarizes them. The example stays offline so dependency handling, execution order, and result contracts remain visible.

Setup#

python -m pip install design-research-agents==0.6.0

Step 1: Define two deterministic handlers#

[1]:

from collections.abc import Sequence

import design_research_agents as agents


def scale_scores(context: agents.WorkflowContext) -> dict[str, object]:
    """Scale the input scores in the first workflow step."""
    scores = context.input_value("scores")
    scale = context.input_value("scale")
    if not isinstance(scores, Sequence) or isinstance(scores, str | bytes):
        raise ValueError("scores must be a sequence")
    if not all(isinstance(score, int | float) for score in scores):
        raise ValueError("scores must contain only numbers")
    if not isinstance(scale, int | float) or isinstance(scale, bool):
        raise ValueError("scale must be numeric")
    return {"scaled_scores": [float(score) * float(scale) for score in scores]}


def summarize_scores(context: agents.WorkflowContext) -> dict[str, object]:
    """Summarize the scaled scores produced by the dependency step."""
    scaled_scores = context.dependency_output("scale_scores").get("scaled_scores")
    if not isinstance(scaled_scores, Sequence) or isinstance(scaled_scores, str | bytes):
        raise ValueError("Scaled scores are missing")
    if not all(isinstance(score, int | float) for score in scaled_scores):
        raise ValueError("Scaled scores must contain only numbers")
    return {
        "count": len(scaled_scores),
        "mean": sum(float(score) for score in scaled_scores) / len(scaled_scores),
    }


print("Handlers:", scale_scores.__name__, "->", summarize_scores.__name__)

Handlers: scale_scores -> summarize_scores

Step 2: Assemble an explicit dependency graph#

[2]:

workflow = agents.Workflow(
    input_schema={
        "type": "object",
        "required": ["scores", "scale"],
    },
    steps=(
        agents.LogicStep(step_id="scale_scores", handler=scale_scores),
        agents.LogicStep(
            step_id="summarize_scores",
            dependencies=("scale_scores",),
            handler=summarize_scores,
        ),
    ),
)
print(workflow.to_mermaid(direction="LR"))

flowchart LR
    workflow_entry["Workflow Entrypoint"]
    step_1["scale_scores<br/>LogicStep"]
    step_2["summarize_scores<br/>LogicStep"]
    workflow_entry --> step_1
    step_1 --> step_2

Step 3: Run the DAG#

[3]:

result = workflow.run(
    {"scores": [0.2, 0.5, 0.8], "scale": 10},
    execution_mode="dag",
    request_id="tutorial-agents-workflow",
)
print("Workflow success:", result.success)
print("Execution order:", " -> ".join(result.execution_order))

Workflow success: True
Execution order: scale_scores -> summarize_scores

Step 4: Inspect intermediate and final outputs#

[4]:

scaled = result.step_results["scale_scores"].output_list("scaled_scores")
summary = result.step_results["summarize_scores"]
print("Scaled scores:", scaled)
print("Count:", summary.output_value("count"))
print("Mean:", f"{float(summary.output_value('mean', 0.0)):.1f}")

Scaled scores: [2.0, 5.0, 8.0]
Count: 3
Mean: 5.0

Next steps#

Substitute ToolStep, ModelStep, or DelegateStep only where external behavior is actually needed. Keeping preprocessing and validation in deterministic logic steps makes model-facing work smaller and easier to inspect.