Skip to main content

Challenge 13: Prompt Flow

Estimated Time

45-60 min | Cost: ~$1.00 (estimated) | Domain: Generative AI Solutions (15-20%)

Exam skills covered​

  • Implement a prompt flow solution
  • Create and configure flow types (standard, chat, evaluation)
  • Define flow nodes and connections for orchestration

Overview​

Prompt flow is an orchestration framework within Azure AI Foundry that enables you to build, test, and deploy LLM-powered workflows as directed acyclic graphs (DAGs). Each flow consists of nodes that execute sequentially or in parallel—including LLM nodes (calling Azure OpenAI), Python nodes (custom logic), and Prompt nodes (template rendering). Flows are defined in YAML and can be executed locally during development or deployed as managed endpoints.

There are three primary flow types: Standard flows for general-purpose LLM pipelines (data processing, summarization, extraction), Chat flows optimized for conversational applications with built-in chat history management, and Evaluation flows that assess the quality of other flows using metrics like groundedness, relevance, and coherence. The evaluation flow pattern is critical for production AI systems—it enables systematic quality measurement before deployment.

Variants allow A/B testing of prompts within a single flow. You can define multiple prompt variants for an LLM node and run them against the same inputs to compare quality and cost. Connections link flows to external services (Azure OpenAI, AI Search) and are configured at the workspace level, allowing secure credential management separate from flow logic.

Architecture​

Prompt flow orchestrates LLM calls, Python processing, and prompt templates as connected nodes in a DAG, with connections to external AI services.

Challenge 13 topology

Prerequisites​

  • Azure AI Foundry project (from Challenge 11)
  • Azure OpenAI deployment (from Challenge 12)
  • Python 3.9+ with promptflow and promptflow-tools packages
  • Azure CLI with ml extension

Implementation​

Task 1: Create a Standard Prompt Flow​

import os
from promptflow.client import PFClient
from promptflow.entities import Run

# Initialize the Prompt Flow client (local development)
pf = PFClient()

# Define flow YAML programmatically
flow_yaml = """
$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Flow.schema.json
inputs:
  topic:
    type: string
    default: "Azure AI Services"
outputs:
  summary:
    type: string
    reference: ${summarize.output}
nodes:
  - name: generate_content
    type: llm
    source:
      type: code
      path: generate_content.jinja2
    inputs:
      deployment_name: gpt-4o-standard
      max_tokens: 500
      temperature: 0.7
      topic: ${inputs.topic}
    connection: aoai-connection
    api: chat
  - name: summarize
    type: python
    source:
      type: code
      path: summarize.py
    inputs:
      content: ${generate_content.output}
"""

# Write flow definition
os.makedirs("my-standard-flow", exist_ok=True)
with open("my-standard-flow/flow.dag.yaml", "w") as f:
    f.write(flow_yaml)

# Create the LLM node prompt template
prompt_template = """system:
You are a technical writer specializing in Azure cloud services.

user:
Write a detailed paragraph about: {{topic}}
"""
with open("my-standard-flow/generate_content.jinja2", "w") as f:
    f.write(prompt_template)

# Create the Python processing node
python_node = """
def summarize(content: str) -> str:
    \"\"\"Extract key points from generated content.\"\"\"
    lines = content.strip().split(". ")
    key_points = [line.strip() for line in lines[:3]]
    summary = "Key points: " + "; ".join(key_points)
    return summary
"""
with open("my-standard-flow/summarize.py", "w") as f:
    f.write(python_node)

print("Flow created at: my-standard-flow/")
print("Files: flow.dag.yaml, generate_content.jinja2, summarize.py")

Task 2: Add an Evaluation Flow​

import os
from promptflow.client import PFClient
from promptflow.entities import Run, AzureOpenAIConnection

pf = PFClient()

# Define evaluation flow
eval_flow_yaml = """
$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Flow.schema.json
inputs:
  question:
    type: string
  answer:
    type: string
  ground_truth:
    type: string
outputs:
  relevance_score:
    type: string
    reference: ${evaluate_relevance.output}
nodes:
  - name: evaluate_relevance
    type: llm
    source:
      type: code
      path: evaluate_relevance.jinja2
    inputs:
      deployment_name: gpt-4o-standard
      max_tokens: 50
      temperature: 0.0
      question: ${inputs.question}
      answer: ${inputs.answer}
      ground_truth: ${inputs.ground_truth}
    connection: aoai-connection
    api: chat
"""

os.makedirs("eval-flow", exist_ok=True)
with open("eval-flow/flow.dag.yaml", "w") as f:
    f.write(eval_flow_yaml)

# Evaluation prompt template
eval_prompt = """system:
You are an AI quality evaluator. Score the answer's relevance to the question on a scale of 1-5.
Return ONLY a number.

user:
Question: {{question}}
Answer: {{answer}}
Ground Truth: {{ground_truth}}

Score (1-5):
"""
with open("eval-flow/evaluate_relevance.jinja2", "w") as f:
    f.write(eval_prompt)

# Create test data for evaluation
import json
test_data = [
    {
        "question": "What is Azure AI Foundry?",
        "answer": "Azure AI Foundry is a platform for building generative AI apps.",
        "ground_truth": "Azure AI Foundry is a unified platform for building, testing, and deploying AI solutions with hub and project architecture."
    },
    {
        "question": "What models does Azure OpenAI support?",
        "answer": "Azure OpenAI supports GPT-4o, GPT-4o-mini, and other models.",
        "ground_truth": "Azure OpenAI supports GPT-4o, GPT-4o-mini, GPT-4, DALL-E, Whisper, and text embedding models."
    }
]
with open("eval-flow/test_data.jsonl", "w") as f:
    for item in test_data:
        f.write(json.dumps(item) + "\n")

# Run the evaluation
eval_run = pf.run(
    flow="eval-flow",
    data="eval-flow/test_data.jsonl",
    name="eval-run-001"
)
print(f"Evaluation run: {eval_run.name}")
print(f"Status: {eval_run.status}")

# Get results
details = pf.get_details(eval_run)
print(f"\nResults:\n{details}")

Task 3: Test Flow with Variants​

import os
from promptflow.client import PFClient

pf = PFClient()

# Create a flow with prompt variants for A/B testing
variant_flow_yaml = """
$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Flow.schema.json
inputs:
  topic:
    type: string
    default: "Azure Kubernetes Service"
outputs:
  result:
    type: string
    reference: ${generate.output}
nodes:
  - name: generate
    type: llm
    source:
      type: code
      path: generate.jinja2
    inputs:
      deployment_name: gpt-4o-standard
      max_tokens: 300
      temperature: 0.7
      topic: ${inputs.topic}
    connection: aoai-connection
    api: chat
    use_variants: true
"""

os.makedirs("variant-flow", exist_ok=True)
with open("variant-flow/flow.dag.yaml", "w") as f:
    f.write(variant_flow_yaml)

# Variant 0: Concise style
variant_0 = """system:
You are a concise technical writer. Keep responses under 100 words.

user:
Explain: {{topic}}
"""
with open("variant-flow/generate.jinja2", "w") as f:
    f.write(variant_0)

# Variant 1: Detailed style
variant_1 = """system:
You are a detailed technical writer. Provide comprehensive explanations with examples.

user:
Provide a detailed explanation of: {{topic}}
Include at least one practical example.
"""
with open("variant-flow/generate_variant_1.jinja2", "w") as f:
    f.write(variant_1)

# Test with default variant
result_0 = pf.test(flow="variant-flow", inputs={"topic": "Azure Kubernetes Service"})
print(f"Variant 0 (Concise): {result_0['result'][:100]}...")

# Test with variant 1
result_1 = pf.test(
    flow="variant-flow",
    inputs={"topic": "Azure Kubernetes Service"},
    node="generate",
    variant="${generate.variant_1}"
)
print(f"\nVariant 1 (Detailed): {result_1['result'][:100]}...")

Task 4: Deploy Flow to Managed Endpoint​

from azure.identity import DefaultAzureCredential
from azure.ai.ml import MLClient
from azure.ai.ml.entities import (
    ManagedOnlineEndpoint,
    ManagedOnlineDeployment,
    Model,
    Environment,
)

credential = DefaultAzureCredential()
ml_client = MLClient(
    credential=credential,
    subscription_id="YOUR_SUBSCRIPTION_ID",
    resource_group_name="rg-ai102-challenge13",
    workspace_name="proj-ai102-genai",
)

# Create managed online endpoint
endpoint = ManagedOnlineEndpoint(
    name="pf-standard-flow-endpoint",
    description="Prompt Flow standard flow endpoint",
    auth_mode="key",
)
ml_client.online_endpoints.begin_create_or_update(endpoint).result()
print(f"Endpoint created: {endpoint.name}")

# Deploy the flow
deployment = ManagedOnlineDeployment(
    name="default",
    endpoint_name="pf-standard-flow-endpoint",
    model=Model(path="my-standard-flow"),
    instance_type="Standard_DS3_v2",
    instance_count=1,
)
ml_client.online_deployments.begin_create_or_update(deployment).result()
print(f"Deployment created: {deployment.name}")

# Set traffic to 100%
endpoint.traffic = {"default": 100}
ml_client.online_endpoints.begin_create_or_update(endpoint).result()

# Test the deployed endpoint
import json
result = ml_client.online_endpoints.invoke(
    endpoint_name="pf-standard-flow-endpoint",
    request_file=json.dumps({"topic": "Azure AI Services"}),
)
print(f"Response: {result}")

Expected Output​

After completing all tasks, you should have:

  1. Standard flow (my-standard-flow/) with:
    • flow.dag.yaml defining the DAG structure
    • LLM node using generate_content.jinja2 template
    • Python node with summarize.py logic
  2. Evaluation flow (eval-flow/) that scores relevance 1-5
  3. Variant flow demonstrating A/B testing of prompt styles
  4. Deployed endpoint serving the standard flow with key-based auth

Break & fix​

ScenarioSymptomRoot CauseFix
Flow test failsConnectionNotFound: aoai-connectionConnection not configured locallyRun pf connection create --file connection.yaml or set env vars
Node output emptyDownstream node receives NoneIncorrect ${node.output} referenceVerify node names match exactly in YAML references
Evaluation scores all 1LLM returns unexpected formatPrompt not constrained enoughAdd "Return ONLY a number between 1 and 5" to system message
Deployment failsModelNotFound during deploymentFlow path incorrect or missing dependenciesEnsure requirements.txt is in flow directory
Variant not foundVariantNotFound errorVariant file naming doesn't match conventionUse {node_name}_variant_{N}.jinja2 naming pattern

Knowledge Check​

1. What are the three primary flow types in Prompt Flow?

2. How do variants enable prompt optimization in Prompt Flow?

3. What is the correct syntax to reference a node's output in flow YAML?

4. What is the purpose of a connection in Prompt Flow?

5. Which node type would you use to implement custom data transformation logic in a flow?

Cleanup​

# Delete endpoint (stops billing for compute)
az ml online-endpoint delete \
  --name "pf-standard-flow-endpoint" \
  --resource-group rg-ai102-challenge13 \
  --workspace-name proj-ai102-genai --yes

az group delete --name rg-ai102-challenge13 --yes --no-wait

Learn More​