# Prompt Flow PRS vs. MAF PRS — Side-by-Side Mapping

Use this table during Phase 1 audit to translate each piece of the existing
PF PRS submission into the MAF equivalent.

## 0. Background — what `load_component(flow.dag.yaml)` did automatically

In the Prompt Flow PRS pattern (see
[examples/tutorials/run-flow-with-pipeline/pipeline.ipynb](../../../../examples/tutorials/run-flow-with-pipeline/pipeline.ipynb)),
`load_component("flow.dag.yaml")` produced an Azure ML **parallel component**
with the following pieces filled in for free:

| Auto-generated piece | Where it came from |
|---|---|
| Input port `data` (`uri_file` / `uri_folder`) | Implicit from PRS |
| Output port `flow_outputs` (`uri_file` → `parallel_run_step.jsonl`) | Implicit from PRS |
| Output port `debug_info` (`uri_folder`) | Implicit from PRS |
| Component parameters (flow inputs + connections) | Parsed from `flow.dag.yaml` |
| Environment | Inherited from latest promptflow runtime image |
| Entry script (`init`/`run`) | Generated by promptflow runtime |
| Column mapping (`url="${data.url}"`) | Driven by flow input names |

MAF has **no equivalent auto-conversion**. The skill produces the same
five artefacts by hand (entry script + processor/executor + component YAML
+ conda env + submission script).

## 1. Component artefacts

| Concern | Prompt Flow PRS | MAF PRS (this skill) |
|---|---|---|
| Component definition | Auto-generated by `load_component("flow.dag.yaml")` | Hand-written `component.yaml` (`type: parallel`) |
| Entry script | Provided by promptflow runtime | `src/entry.py` — thin wrapper exposing `init()` / `run(mini_batch, context)` / `shutdown()` |
| Plumbing layer | `promptflow.parallel` (`AbstractParallelRunProcessor` + `ComponentRunExecutor`) | `src/maf_prs/{processor,executor,config}.py` (mirrors the same split) |
| Environment | Inherited from latest promptflow runtime image | `env/conda.yml` declared in the component |
| Component parameters | Auto-derived from flow inputs + `connections` | Declared explicitly under `inputs:` in `component.yaml` |
| Input port type | PF accepted `uri_file` directly (runtime emitted the `--amlbi_pf_*` flag set automatically) | Vanilla PRS rejects `uri_file` unless `program_arguments` carries the same PF compatibility flag set (`--amlbi_pf_enabled True --amlbi_pf_run_mode component --amlbi_file_format jsonl --amlbi_mini_batch_rows 1`). Default for this skill (gotcha #12). |
| Output ports | `flow_outputs` (jsonl), `debug_info` (folder) | Same names, declared explicitly |
| Append rule | `parallel_run_step.jsonl` | `append_row_to: ${{outputs.flow_outputs}}` |

## 2. Per-row data binding

| Concern | Prompt Flow PRS | MAF PRS |
|---|---|---|
| Column → input mapping | `flow_node(url="${data.url}")` declared in pipeline DSL; resolved at runtime by `FlowExecutor.apply_inputs_mapping` | Pure Python: `hooks.build_workflow_input(row)` reads `row["url"]` and returns whatever the workflow's first executor expects |
| Where the mapping lives | `submit_pipeline.py` (declarative) | `src/hooks.py` (imperative) — the only file most users edit |
| Input format | `PF_INPUT_FORMAT` env var | `processor.py` uses `pandas`; jsonl/csv/tsv all work without an env var |
| File-mode input (`uri_folder` of opaque files) | PF iterates files of allowed extensions | `processor._iter_rows()` yields `{"path": ...}` per file |
| Stable row id | `Row.from_dict(data, row_number=base+idx)` where `base = context.global_row_index_lower_bound` | Same: `processor.process()` reads `context.global_row_index_lower_bound` and stamps `line_number` on each result |

## 3. Connections / secrets

| Concern | Prompt Flow PRS | MAF PRS |
|---|---|---|
| Endpoint URL + deployment | `connections={"node": {"connection": "...", "deployment_name": "..."}}` | Component `inputs:` (e.g. `model_endpoint`, `model_deployment`) wired through `program_arguments` to env vars |
| API key | Stored in PF connection | Prefer **Managed Identity + Key Vault**; if a key is unavoidable, inject as a workspace secret env var on the deployment, never in YAML |
| Multiple LLM nodes with different connections | Per-node `connections` map | One set of inputs per distinct chat client; usually a single `(endpoint, deployment)` pair suffices because the workflow already encapsulates routing |

## 4. PRS run settings (carry over verbatim)

These have a 1:1 mapping — copy the values from the original PF script
unchanged unless the user explicitly wants to tune.

| PF (`flow_node.*`) | MAF (`component.yaml` and `pipeline_node.*`) |
|---|---|
| `compute = "cpu-cluster"` | `pipeline_node.compute = "cpu-cluster"` |
| `resources = {"instance_count": N}` | `pipeline_node.resources = {"instance_count": N}` |
| `mini_batch_size = K` | `pipeline_node.mini_batch_size = K` (and default in `component.yaml`) |
| `max_concurrency_per_instance = M` | same |
| `retry_settings = {"max_retries": ..., "timeout": ...}` | same |
| `error_threshold = -1` | same |
| `mini_batch_error_threshold = -1` | same |
| `logging_level = "DEBUG"` | same |
| `environment_variables = {"PF_INPUT_FORMAT": "jsonl"}` | Pass via `program_arguments` or `pipeline_node.environment_variables` |
| `outputs.flow_outputs.mode = "mount"` (when `instance_count > 1`) | same — required, not optional |
| `outputs.debug_info.mode = "mount"` (when `instance_count > 1`) | same |

## 5. Pipeline DSL

| Concern | Prompt Flow PRS | MAF PRS |
|---|---|---|
| Imports | `from azure.ai.ml import load_component, MLClient, Input, Output` | Same |
| `@pipeline()` decorator | `from azure.ai.ml.dsl import pipeline` | Same |
| Column-mapping arguments | `flow_node(url="${data.url}", question="${data.q}")` | **Not present** — mapping moved into Python (`executor.build_workflow_input`); the pipeline call only passes `data` + connection inputs |
| Submission | `ml_client.jobs.create_or_update(pipeline_job_def, experiment_name=...)` | Same |
| Streaming | `ml_client.jobs.stream(...)` | Same |
| Scheduler / batch endpoint (notebook §4) | works on PF flow_component | works unchanged on the MAF parallel component — no special handling needed |

## 6. Things PF did automatically that you must do explicitly

| Auto-done by PF | You must do this in MAF |
|---|---|
| Run a forked Python process per worker that hosts the flow runtime | `entry.py` exposes `init()` / `run(mini_batch, context)` / `shutdown()`; `processor.init()` builds a workflow factory + reusable event loop |
| Convert each row into a flow run | `executor.execute(row, row_number)` builds the input via `hooks.build_workflow_input(row)` and `await`s `workflow.run(...)` |
| Apply `${data.col}` template mapping | Edit `hooks.build_workflow_input(row)` in `src/hooks.py` — plain Python, no template engine. The skill agent fills this automatically when the source PF mapping is parseable **and** the workflow's start handler input is typed (see SKILL.md Phase 1.5 checks A–D); otherwise it leaves a `# TODO` stub naming the missing piece. |
| Append `parallel_run_step.jsonl` lines | `processor.process(...)` returns a `list[str]` from JSON-serialised dicts; PRS appends each line |
| Surface `debug_info` automatically | Decide what (if anything) to write to `--output_dir` in `hooks.setup` / a custom `executor.finalize` |
| Validate input/output ports | Validate manually by running `entry.py` against the local sample before submitting |
| Aggregation node finalize (`AggregationFinalizer`) | Override `executor.finalize()` to consume the temp jsonl rows if you need a global reduce step |
