Prompt Workflow Agent#

Source: examples/agents/prompt_workflow_agent.py

Introduction#

This example shows how to package a prompt-mode Workflow as a reusable study agent for deterministic design experiments. PromptWorkflowAgent keeps the workflow itself simple while moving packaged-problem, run-spec, and condition formatting into one explicit prompt builder.

Technical Implementation#

  1. Define tiny local study packet dataclasses so the example stays dependency-light and deterministic.

  2. Build a prompt-mode Workflow with logic steps that capture the resolved study prompt and emit one final summary payload.

  3. Wrap that workflow in PromptWorkflowAgent with a prompt builder that converts study metadata into one canonical prompt string.

  4. Run the delegate with a fixed request_id and print a compact JSON payload for docs and regression tests.

        flowchart LR
    A["Problem packet + run spec + condition"] --> B["PromptWorkflowAgent(prompt_builder)"]
    B --> C["Prompt-mode Workflow"]
    C --> D["capture_study_prompt"]
    D --> E["emit_summary"]
    E --> F["JSON payload"]
    
  1from __future__ import annotations
  2
  3import json
  4from dataclasses import dataclass
  5from typing import Any
  6
  7import design_research_agents as drag
  8
  9WORKFLOW_DIAGRAM_DIRECTION = "LR"
 10
 11
 12@dataclass(frozen=True)
 13class _ProblemPacket:
 14    problem_id: str
 15    brief: str
 16
 17
 18@dataclass(frozen=True)
 19class _RunSpec:
 20    run_id: str
 21    objective: str
 22
 23
 24@dataclass(frozen=True)
 25class _Condition:
 26    condition_id: str
 27    budget_label: str
 28
 29
 30def _capture_study_prompt(context: dict[str, Any]) -> dict[str, object]:
 31    """Capture the resolved study prompt and request id inside the workflow."""
 32    workflow_metadata = context.get("_workflow")
 33    request_id = ""
 34    if isinstance(workflow_metadata, dict):
 35        raw_request_id = workflow_metadata.get("request_id")
 36        if isinstance(raw_request_id, str):
 37            request_id = raw_request_id
 38    return {
 39        "study_prompt": str(context["prompt"]),
 40        "request_id": request_id,
 41    }
 42
 43
 44def _emit_summary(context: dict[str, Any]) -> dict[str, object]:
 45    """Emit the canonical final_output payload for this example run."""
 46    capture_output = context["dependency_results"]["capture_study_prompt"]["output"]
 47    return {
 48        "final_output": {
 49            "request_id": capture_output["request_id"],
 50            "study_prompt": capture_output["study_prompt"],
 51            "workflow_step": str(context["_workflow"]["step_id"]),
 52        }
 53    }
 54
 55
 56def build_example_workflow() -> drag.Workflow:
 57    """Build the prompt-mode workflow wrapped by the prompt workflow agent."""
 58    return drag.Workflow(
 59        steps=[
 60            drag.LogicStep(
 61                step_id="capture_study_prompt",
 62                handler=_capture_study_prompt,
 63            ),
 64            drag.LogicStep(
 65                step_id="emit_summary",
 66                dependencies=("capture_study_prompt",),
 67                handler=_emit_summary,
 68            ),
 69        ]
 70    )
 71
 72
 73def _build_study_prompt(problem_packet: object, run_spec: object, condition: object) -> str:
 74    """Translate study packet objects into one deterministic workflow prompt."""
 75    if not isinstance(problem_packet, _ProblemPacket):
 76        raise TypeError("Expected _ProblemPacket for problem_packet.")
 77    if not isinstance(run_spec, _RunSpec):
 78        raise TypeError("Expected _RunSpec for run_spec.")
 79    if not isinstance(condition, _Condition):
 80        raise TypeError("Expected _Condition for condition.")
 81    return (
 82        f"Problem: {problem_packet.problem_id}. Brief: {problem_packet.brief} "
 83        f"Run: {run_spec.run_id}. Objective: {run_spec.objective}. "
 84        f"Condition: {condition.condition_id} ({condition.budget_label})."
 85    )
 86
 87
 88def main() -> None:
 89    """Run the workflow-backed study agent and print a deterministic summary payload."""
 90    agent = drag.PromptWorkflowAgent(
 91        workflow=build_example_workflow(),
 92        prompt_builder=_build_study_prompt,
 93    )
 94    problem_packet = _ProblemPacket(
 95        problem_id="cooling_plate_redesign",
 96        brief="Reduce pressure drop while preserving manufacturability.",
 97    )
 98    run_spec = _RunSpec(
 99        run_id="study-run-12",
100        objective="Summarize the design-study setup for a control workflow.",
101    )
102    condition = _Condition(
103        condition_id="control_workflow",
104        budget_label="single-pass",
105    )
106    result = agent.run(
107        "Fallback prompts are also supported, but this example uses study dependencies.",
108        request_id="example-prompt-workflow-agent-001",
109        dependencies={
110            "problem_packet": problem_packet,
111            "run_spec": run_spec,
112            "condition": condition,
113        },
114    )
115    payload = {
116        "workflow_mermaid": agent.workflow.to_mermaid(direction=WORKFLOW_DIAGRAM_DIRECTION),
117        "summary": result.summary(),
118    }
119    print(json.dumps(payload, ensure_ascii=True, indent=2, sort_keys=True))
120
121
122if __name__ == "__main__":
123    main()

Expected Results#

Run Command

PYTHONPATH=src python3 examples/agents/prompt_workflow_agent.py

Example output shape:

{
  "workflow_mermaid": "flowchart LR ...",
  "summary": {
    "success": true,
    "final_output": {
     "request_id": "example-prompt-workflow-agent-001",
      "study_prompt": "Problem: cooling_plate_redesign...",
      "workflow_step": "emit_summary"
    },
    "terminated_reason": null,
    "error": null,
    "trace": {
      "request_id": "example-workflow-study-delegate-001"
    }
  }
}

References#