Complete Prompt Flow Tutorial: Microsoft's Framework for Building LLM Applications

Prompt Flow is an open-source framework from Microsoft designed to streamline the development, testing, and deployment of applications built on Large Language Models (LLMs). The framework provides a flow-based approach that enables developers to build structured, testable, and production-ready LLM pipelines.

In this tutorial, we will learn how to use Prompt Flow from installation to deployment, complete with runnable code examples.

Why Prompt Flow?

Building reliable LLM applications requires more than just calling an API. You need to manage prompts, connect various components, evaluate output quality, and ensure consistent application behavior. Prompt Flow addresses these challenges by providing:

• Flow-based development: Build LLM pipelines as clear, structured DAGs (Directed Acyclic Graphs)
• Built-in evaluation: Tools for systematically measuring LLM output quality
• Tracing and debugging: Capabilities to track every execution step for efficient debugging
• CI/CD integration: Support for integrating flows into DevOps pipelines
• Multi-provider support: Compatible with OpenAI, Azure OpenAI, and other LLM models

Installation

Prerequisites

Ensure Python 3.9 or later is installed on your system.

python --version

Installing Prompt Flow

Install the main Prompt Flow package along with its supporting tools:

pip install promptflow promptflow-tools promptflow-devkit

To use Azure OpenAI as the LLM backend:

pip install promptflow-azure

Verify installation:

pf --version

Configuring Connections

Prompt Flow uses the concept of "connections" to securely manage API credentials. Let's create a connection for OpenAI:

pf connection create --file connectionopenai.yaml

Create the connectionopenai.yaml file:

$schema: https://azuremlschemas.azureedge.net/promptflow/latest/OpenAIConnection.schema.json
name: openaiconnection
type: openai

apikey: ""
organization: ""

For Azure OpenAI:

$schema: https://azuremlschemas.azureedge.net/promptflow/latest/AzureOpenAIConnection.schema.json
name: azureopenaiconnection
type: azureopenai
apikey: ""

apibase: "https://.openai.azure.com/"
apiversion: "2024-02-01"

pf connection list

mkdir myfirstflow
cd myfirstflow

$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Flow.schema.json
inputs:
  question:
    type: string
    default: "What is machine learning?"
outputs:
  answer:
    type: string
    reference: ${summarize.output}
nodes:
name: classifyintent
    type: python

    source:
      type: code
      path: classifyintent.py

    inputs:
      question: ${inputs.question}
name: generateanswer
    type: llm

    source:
      type: code
      path: generateanswer.jinja2

    inputs:
      deploymentname: gpt-4
      question: ${inputs.question}
      intent: ${classifyintent.output}

    connection: openaiconnection

    api: chat
name: summarize
    type: python

    source:
      type: code
      path: summarize.py
    inputs:
      answer: ${generateanswer.output}

Create the classifyintent.py file:

from promptflow.core import tool

@tool
def classifyintent(question: str) -> str:
    """Classify the intent of a user question."""
    questionlower = question.lower()


    if any(word in questionlower for word in ["how", "tutorial", "guide", "step"]):
        return "tutorial"
    elif any(word in questionlower for word in ["what", "explain", "describe", "define"]):

        return "explanation"
    elif any(word in questionlower for word in ["compare", "vs", "versus", "difference"]):
        return "comparison"
    else:
        return "general"

Create the generateanswer.jinja2 prompt template:

system:
You are an AI assistant specializing in technology and data science.
Provide informative and easy-to-understand answers.

Question type: {{intent}}

user:
{{question}}

from promptflow.core import tool

@tool
def summarize(answer: str) -> str:
    """Summarize and format the LLM answer."""
    if len(answer) > 500:
        return answer[:500] + "..."
    return answer

pf flow test --flow . --detail

pf flow test --flow . --inputs question="How do I deploy an ML model to production?"

{"question": "What is a neural network?"}
{"question": "How do I fine-tune an LLM?"}
{"question": "Compare PyTorch and TensorFlow"}
{"question": "Explain the concept of transfer learning"}

pf run create --flow . --data data.jsonl --name batchtestrun --stream

pf run show-details --name batchtestrun

Create the chatflow.py file:

from dataclasses import dataclass
from promptflow.tracing import trace
from promptflow.core import AzureOpenAIModelConfiguration, OpenAIModelConfiguration

@dataclass
class ChatFlow:
    modelconfig: OpenAIModelConfiguration


    def init(self, modelconfig: OpenAIModelConfiguration):
        self.modelconfig = modelconfig
        self.chathistory = []


    @trace
    def call(self, question: str) -> str:
        from openai import OpenAI

        client = OpenAI(
            apikey=self.modelconfig.apikey,
        )

        systemmessage = {

            "role": "system",
            "content": (
                "You are an AI assistant that helps answer questions "
                "about data science and machine learning. "
                "Answer in clear, understandable language."
            )
        }

        messages = [systemmessage] + self.chathistory + [

            {"role": "user", "content": question}
        ]

        response = client.chat.completions.create(
            model=self.modelconfig.model,
            messages=messages,
            temperature=0.7,
            maxtokens=1000

        )

        answer = response.choices[0].message.content
        self.chathistory.append({"role": "user", "content": question})
        self.chathistory.append({"role": "assistant", "content": answer})


        return answer


if name == "main":
    config = OpenAIModelConfiguration(
        model="gpt-4",
        apikey="your-api-key"
    )
    flow = ChatFlow(modelconfig=config)

    result = flow("What is RAG in the context of LLMs?")
    print(result)

$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Flow.schema.json
entry: chatflow:ChatFlow
init:
  modelconfig:

    type: OpenAIModelConfiguration
    model: gpt-4
    apikey: ${openaiconnection.apikey}

from promptflow.tracing import trace
from promptflow.core import tool

@trace
def simpleqa(question: str, context: str = "") -> dict:
    """Simple question-answering flow."""
    from openai import OpenAI
    import os

    client = OpenAI(apikey=os.environ.get("OPENAIAPIKEY"))


    prompt = f"""Based on the following context, answer the given question.

Context: {context if context else 'No specific context provided.'}

Question: {question}

Answer:"""

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3
    )

    answer = response.choices[0].message.content

    return {
        "answer": answer,
        "tokensused": response.usage.totaltokens,

        "model": response.model
    }

Create the evalflow.py file:

from promptflow.core import tool
from promptflow.evals.evaluators import (
    RelevanceEvaluator,
    CoherenceEvaluator,
    GroundednessEvaluator
)

@tool
def evaluateanswer(

    question: str,
    answer: str,
    groundtruth: str = ""
) -> dict:
    """Evaluate LLM answer quality."""

    scores = {}

    relevancescore = calculaterelevance(question, answer)
    scores["relevance"] = relevancescore


    coherencescore = calculatecoherence(answer)

    scores["coherence"] = coherencescore

    if groundtruth:

        accuracyscore = calculateaccuracy(answer, groundtruth)
        scores["accuracy"] = accuracyscore


    scores["overall"] = sum(scores.values()) / len(scores)

    return scores


def calculaterelevance(question: str, answer: str) -> float:
    """Calculate relevance score of the answer to the question."""
    questionwords = set(question.lower().split())

    answerwords = set(answer.lower().split())
    overlap = questionwords.intersection(answerwords)
    if not questionwords:

        return 0.0
    return min(len(overlap) / len(questionwords), 1.0)


def calculatecoherence(text: str) -> float:

    """Calculate text coherence score."""
    sentences = text.split(".")
    if len(sentences) <= 1:
        return 0.5
    return min(len(sentences) / 10, 1.0)


def calculateaccuracy(answer: str, groundtruth: str) -> float:

    """Calculate accuracy score against ground truth."""
    answerwords = set(answer.lower().split())
    truthwords = set(groundtruth.lower().split())
    if not truthwords:

        return 0.0
    overlap = answerwords.intersection(truthwords)

    return len(overlap) / len(truthwords)

Using Built-in Evaluators

Prompt Flow provides built-in evaluators that use LLMs to assess output quality:

from promptflow.evals.evaluate import evaluate
from promptflow.evals.evaluators import (
    RelevanceEvaluator,
    GroundednessEvaluator,
    FluencyEvaluator,
    CoherenceEvaluator,
    SimilarityEvaluator
)
from promptflow.core import AzureOpenAIModelConfiguration

modelconfig = AzureOpenAIModelConfiguration(

    azuredeployment="gpt-4",
    apikey="your-api-key",

    azureendpoint="https://your-resource.openai.azure.com/"
)

relevanceevaluator = RelevanceEvaluator(modelconfig=modelconfig)

coherenceevaluator = CoherenceEvaluator(modelconfig=modelconfig)

result = evaluate(
    data="testdata.jsonl",

    evaluators={
        "relevance": relevanceevaluator,
        "coherence": coherenceevaluator,

    },
    evaluatorconfig={
        "relevance": {
            "question": "${data.question}",
            "answer": "${data.answer}",
            "context": "${data.context}"
        },
        "coherence": {
            "question": "${data.question}",
            "answer": "${data.answer}"
        }
    }
)

print(f"Metrics: {result['metrics']}")
print(f"Row-level results: {result['rows']}")

Tracing for Debugging

Prompt Flow provides automatic tracing to monitor flow execution:

from promptflow.tracing import trace, starttrace
import json

starttrace()

@trace
def processdocument(document: str) -> dict:

    """Process a document with tracing."""

    chunks = chunkdocument(document)
    embeddings = generateembeddings(chunks)

    summary = generatesummary(chunks)

    return {
        "numchunks": len(chunks),

        "summary": summary,
        "embeddingdim": len(embeddings[0]) if embeddings else 0
    }

@trace
def chunkdocument(document: str, chunksize: int = 500) -> list:
    """Split document into chunks."""
    words = document.split()
    chunks = []
    for i in range(0, len(words), chunksize):

        chunk = " ".join(words[i:i + chunksize])
        chunks.append(chunk)
    return chunks

@trace
def generateembeddings(chunks: list) -> list:

    """Generate embeddings for each chunk."""
    from openai import OpenAI
    import os

    client = OpenAI(apikey=os.environ.get("OPENAIAPIKEY"))
    embeddings = []

    for chunk in chunks:
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=chunk
        )
        embeddings.append(response.data[0].embedding)

    return embeddings

@trace
def generatesummary(chunks: list) -> str:

    """Generate summary from chunks."""
    from openai import OpenAI
    import os

    client = OpenAI(apikey=os.environ.get("OPENAIAPIKEY"))
    combined = " ".join(chunks[:3])

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "Create a brief summary of the following text."},
            {"role": "user", "content": combined}
        ]
    )

    return response.choices[0].message.content

Launch the trace viewer:

pf service start

Access the trace viewer in your browser at http://localhost:23333.

Building a RAG Flow

One of the most common use cases is building a Retrieval-Augmented Generation (RAG) pipeline:

from promptflow.core import tool
from promptflow.tracing import trace
import numpy as np

class RAGFlow:
    def init(self, modelconfig, vectorstorepath: str = "vectors.npz"):

        self.modelconfig = modelconfig

        self.vectorstorepath = vectorstorepath

        self.documents = []
        self.embeddings = []

    @trace
    def indexdocuments(self, documents: list[str]):
        """Index documents into the vector store."""
        from openai import OpenAI

        client = OpenAI(apikey=self.modelconfig.apikey)

        self.documents = documents

        for doc in documents:
            response = client.embeddings.create(
                model="text-embedding-3-small",
                input=doc
            )
            self.embeddings.append(response.data[0].embedding)

        np.savez(
            self.vectorstorepath,

            embeddings=np.array(self.embeddings),
            documents=np.array(documents)
        )

    @trace
    def retrieve(self, query: str, topk: int = 3) -> list[str]:
        """Retrieve relevant documents based on query."""
        from openai import OpenAI

        client = OpenAI(apikey=self.modelconfig.apikey)


        queryresponse = client.embeddings.create(
            model="text-embedding-3-small",
            input=query
        )
        queryembedding = np.array(queryresponse.data[0].embedding)

        embeddings = np.array(self.embeddings)
        similarities = np.dot(embeddings, queryembedding) / (

            np.linalg.norm(embeddings, axis=1) * np.linalg.norm(queryembedding)
        )

        topindices = np.argsort(similarities)[-topk:][::-1]
        return [self.documents[i] for i in topindices]


    @trace
    def generate(self, query: str, context: list[str]) -> str:
        """Generate answer based on retrieved context."""
        from openai import OpenAI

        client = OpenAI(apikey=self.modelconfig.apikey)

        contexttext = "\n\n".join(context)


        response = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "Answer the question based on the provided context. "
                        "If the answer is not in the context, state that the "
                        "information is not available."
                    )
                },
                {
                    "role": "user",
                    "content": f"Context:\n{contexttext}\n\nQuestion: {query}"
                }
            ],
            temperature=0.3
        )

        return response.choices[0].message.content

    @trace
    def call(self, query: str) -> dict:
        """Run the full RAG pipeline."""
        relevantdocs = self.retrieve(query)

        answer = self.generate(query, relevantdocs)
        return {
            "answer": answer,
            "sources": relevantdocs,

            "numsources": len(relevantdocs)

        }

pf flow build --source . --output dist --format docker

cd dist
docker build -t my-flow-app .
docker run -p 8080:8080 -e OPENAICONNECTIONAPIKEY=your-key my-flow-app

pfazure flow create --flow . --subscription  --resource-group  --workspace 

pfazure run create --flow . --data data.jsonl --subscription  --resource-group  --workspace  --stream

pf flow serve --source . --port 8080

curl -X POST http://localhost:8080/score \
  -H "Content-Type: application/json" \
  -d '{"question": "What is prompt engineering?"}'

myflowproject/
├── flows/
│   ├── chatflow/
│   │   ├── flow.dag.yaml
│   │   ├── chat.py
│   │   └── chat.jinja2
│   └── evalflow/

│       ├── flow.dag.yaml
│       └── evaluate.py
├── data/
│   ├── train.jsonl
│   └── test.jsonl
├── connections/
│   └── openai.yaml
├── tests/
│   └── testflow.py
└── requirements.txt

2. Prompt Management

Separate prompts from code logic for easier iteration:

from pathlib import Path
from promptflow.core import tool

PROMPTSDIR = Path(file).parent / "prompts"


@tool
def getprompt(intent: str, language: str = "en") -> str:
    """Load prompt template based on intent and language."""
    promptfile = PROMPTSDIR / f"{intent}{language}.jinja2"

    if promptfile.exists():
        return promptfile.readtext()
    return (PROMPTSDIR / f"default{language}.jinja2").readtext()

from promptflow.core import tool
from tenacity import retry, stopafterattempt, waitexponential

@tool
@retry(
    stop=stopafterattempt(3),
    wait=waitexponential(multiplier=1, min=2, max=10)

)
def callllmwithretry(prompt: str, model: str = "gpt-4") -> str:
    """Call LLM with retry logic."""
    from openai import OpenAI
    import os

    client = OpenAI(apikey=os.environ.get("OPENAIAPIKEY"))


    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7
    )

    return response.choices[0].message.content

import pytest
from promptflow.core import AzureOpenAIModelConfiguration

def testclassifyintent():

    from classifyintent import classifyintent

    assert classifyintent("How do I deploy?") == "tutorial"
    assert classifyintent("What is ML?") == "explanation"

    assert classifyintent("Compare A and B") == "comparison"

def testflowoutput():
    from promptflow.client import PFClient

    pf = PFClient()
    result = pf.test(
        flow=".",
        inputs={"question": "What is a neural network?"}
    )
    assert "answer" in result
    assert len(result["answer"]) > 0

5. Monitoring and Observability

Integrate with Application Insights for production monitoring:

from promptflow.tracing import trace
from opentelemetry import trace as oteltrace

from opentelemetry.exporter.otlp.proto.grpc.traceexporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

provider = TracerProvider()
processor = BatchSpanProcessor(
    OTLPSpanExporter(endpoint="http://localhost:4317")
)
provider.addspanprocessor(processor)
oteltrace.settracer_provider(provider)

Conclusion

Prompt Flow is an invaluable framework for building structured, production-ready LLM applications. With its flow-based approach, developers can build pipelines that are clear, easy to debug, and systematically evaluable.

The key strengths of Prompt Flow include ease of defining flows as DAGs, a built-in evaluation framework, comprehensive tracing, and strong integration with the Azure ecosystem. This framework is ideal for teams looking to build LLM applications with high engineering standards.

For next steps, try building more complex flows such as multi-agent systems or RAG pipelines with automated evaluation. The official Prompt Flow documentation on GitHub and Microsoft Learn provides extensive examples and additional references.