Prompt Workflow Agent#

Source: examples/agents/prompt_workflow_agent.py

Introduction#

This example shows how to package a JSON prompt-mode Workflow as a reusable study agent for deterministic design experiments. PromptWorkflowAgent keeps study prompt construction separate from the workflow that turns model output into one structured JSON payload.

Technical Implementation#

  1. Define tiny local study packet dataclasses plus a public StudyCondition.

  2. Build a prompt-mode Workflow with build_json_prompt_workflow(...) and a tiny deterministic LLM client.

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

  4. Execute an AgentRunRequest through execute_agent_request(...) 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["build_json_prompt_workflow"]
    C --> D["json_response"]
    D --> E["JSON payload"]
    
  1from __future__ import annotations
  2
  3import json
  4from dataclasses import dataclass
  5
  6import design_research_agents as drag
  7
  8WORKFLOW_DIAGRAM_DIRECTION = "LR"
  9
 10
 11@dataclass(frozen=True)
 12class _ProblemPacket:
 13    problem_id: str
 14    brief: str
 15
 16
 17@dataclass(frozen=True)
 18class _RunSpec:
 19    run_id: str
 20    objective: str
 21
 22
 23class _DeterministicJSONClient:
 24    """Small deterministic LLM client used to keep the example offline."""
 25
 26    def generate(self, request: drag.LLMRequest) -> drag.LLMResponse:
 27        """Return a valid JSON response echoing the study prompt."""
 28        prompt = request.messages[-1].content
 29        return drag.LLMResponse(
 30            text=json.dumps(
 31                {
 32                    "study_prompt": prompt,
 33                    "workflow_step": "json_response",
 34                    "prompt_character_count": len(prompt),
 35                },
 36                ensure_ascii=True,
 37                sort_keys=True,
 38            ),
 39            model="deterministic-json-client",
 40            provider="example",
 41            usage={"prompt_tokens": 42, "completion_tokens": 24},
 42        )
 43
 44
 45def build_example_workflow() -> drag.Workflow:
 46    """Build the prompt-mode workflow wrapped by the prompt workflow agent."""
 47    return drag.build_json_prompt_workflow(
 48        llm_client=_DeterministicJSONClient(),
 49        response_schema={
 50            "type": "object",
 51            "properties": {
 52                "study_prompt": {"type": "string"},
 53                "workflow_step": {"type": "string"},
 54                "prompt_character_count": {"type": "number"},
 55            },
 56            "required": ["study_prompt", "workflow_step", "prompt_character_count"],
 57        },
 58        request_metadata={"example": "prompt_workflow_agent"},
 59        default_request_id_prefix="example-prompt-workflow-agent",
 60    )
 61
 62
 63def _build_study_prompt(problem_packet: object, run_spec: object, condition: object) -> str:
 64    """Translate study packet objects into one deterministic workflow prompt."""
 65    if not isinstance(problem_packet, _ProblemPacket):
 66        raise TypeError("Expected _ProblemPacket for problem_packet.")
 67    if not isinstance(run_spec, _RunSpec):
 68        raise TypeError("Expected _RunSpec for run_spec.")
 69    if not isinstance(condition, drag.StudyCondition):
 70        raise TypeError("Expected StudyCondition for condition.")
 71    budget_label = str(condition.metadata.get("budget_label", "unspecified"))
 72    return (
 73        f"Problem: {problem_packet.problem_id}. Brief: {problem_packet.brief} "
 74        f"Run: {run_spec.run_id}. Objective: {run_spec.objective} "
 75        f"Condition: {condition.condition_id} ({budget_label})."
 76    )
 77
 78
 79def main() -> None:
 80    """Run the workflow-backed study agent and print a deterministic summary payload."""
 81    agent = drag.PromptWorkflowAgent(
 82        workflow=build_example_workflow(),
 83        prompt_builder=_build_study_prompt,
 84    )
 85    problem_packet = _ProblemPacket(
 86        problem_id="cooling_plate_redesign",
 87        brief="Reduce pressure drop while preserving manufacturability.",
 88    )
 89    run_spec = _RunSpec(
 90        run_id="study-run-12",
 91        objective="Summarize the design-study setup for a control workflow.",
 92    )
 93    condition = drag.StudyCondition(
 94        condition_id="control_workflow",
 95        label="Control workflow",
 96        metadata={"budget_label": "single-pass"},
 97    )
 98    run_request = drag.AgentRunRequest(
 99        agent_ref=agent,
100        prompt="Fallback prompts are also supported, but this example uses study dependencies.",
101        request_id="example-prompt-workflow-agent-001",
102        dependencies={
103            "problem_packet": problem_packet,
104            "run_spec": run_spec,
105            "condition": condition,
106        },
107    )
108    execution: drag.AgentExecutionEnvelope = drag.execute_agent_request(run_request)
109    compatibility_execution = drag.execute_agent_run(
110        lambda prompt: {"output": {"text": prompt}, "metadata": {"path": "execute_agent_run"}},
111        prompt="Compatibility execution path.",
112        request_id="example-prompt-workflow-agent-compat",
113        dependencies={},
114    )
115    normalized_preview = drag.normalize_agent_execution(
116        {"text": "Normalized execution preview."},
117        request_id="example-prompt-workflow-agent-normalize",
118    )
119    payload = {
120        "workflow_mermaid": agent.workflow.to_mermaid(direction=WORKFLOW_DIAGRAM_DIRECTION),
121        "execution": {
122            "output": execution.output,
123            "metrics": execution.metrics,
124            "trace_refs": execution.trace_refs,
125            "metadata": execution.metadata,
126            "event_count": len(execution.events),
127        },
128        "compatibility": {
129            "output": compatibility_execution.output,
130            "normalized_preview": normalized_preview.output,
131        },
132    }
133    print(json.dumps(payload, ensure_ascii=True, indent=2, sort_keys=True))
134
135
136if __name__ == "__main__":
137    main()

Expected Results#

Run Command

PYTHONPATH=src python3 examples/agents/prompt_workflow_agent.py

Example output shape:

{
  "workflow_mermaid": "flowchart LR ...",
  "execution": {
    "output": {
      "study_prompt": "Problem: cooling_plate_redesign...",
      "workflow_step": "json_response"
    },
    "metrics": {},
    "event_count": 1
  }
}

References#