Tutorial Lengkap Prompt Flow: Framework Microsoft untuk Membangun Aplikasi LLM

Prompt Flow adalah framework open-source dari Microsoft yang dirancang untuk memudahkan pengembangan, pengujian, dan deployment aplikasi berbasis Large Language Model (LLM). Framework ini menyediakan pendekatan berbasis flow yang memungkinkan developer membangun pipeline LLM yang terstruktur, dapat diuji, dan siap produksi.

Dalam tutorial ini, kita akan mempelajari cara menggunakan Prompt Flow dari instalasi hingga deployment, lengkap dengan contoh kode yang bisa langsung dipraktekkan.

Mengapa Prompt Flow?

Membangun aplikasi LLM yang reliable membutuhkan lebih dari sekadar memanggil API. Kita perlu mengelola prompt, menghubungkan berbagai komponen, mengevaluasi kualitas output, dan memastikan aplikasi berjalan konsisten. Prompt Flow menyelesaikan masalah ini dengan menyediakan:

• Flow-based development: Membangun pipeline LLM sebagai DAG (Directed Acyclic Graph) yang jelas dan terstruktur
• Built-in evaluation: Tools untuk mengukur kualitas output LLM secara sistematis
• Tracing dan debugging: Kemampuan melacak setiap langkah eksekusi untuk debugging yang efisien
• CI/CD integration: Dukungan untuk mengintegrasikan flow ke dalam pipeline DevOps
• Multi-provider support: Kompatibel dengan OpenAI, Azure OpenAI, dan model LLM lainnya

Instalasi

Prasyarat

Pastikan Python 3.9 atau lebih baru sudah terinstal di sistem Anda.

python --version

Instalasi Prompt Flow

Instal package utama Prompt Flow beserta tools pendukungnya:

pip install promptflow promptflow-tools promptflow-devkit

Untuk menggunakan Azure OpenAI sebagai backend LLM:

pip install promptflow-azure

Verifikasi instalasi:

pf --version

Konfigurasi Connection

Prompt Flow menggunakan konsep "connection" untuk mengelola kredensial API secara aman. Mari buat connection untuk OpenAI:

pf connection create --file connectionopenai.yaml

Buat file connectionopenai.yaml:

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

apikey: ""
organization: ""

Untuk 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: "Apa itu 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}

Buat file classifyintent.py:

from promptflow.core import tool

@tool
def classifyintent(question: str) -> str:
    """Klasifikasi intent dari pertanyaan user."""
    questionlower = question.lower()


    if any(word in questionlower for word in ["cara", "bagaimana", "how", "tutorial"]):
        return "tutorial"
    elif any(word in questionlower for word in ["apa", "what", "jelaskan", "explain"]):

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

Buat file generateanswer.jinja2 sebagai prompt template:

system:
Kamu adalah asisten AI yang ahli dalam bidang teknologi dan data science.
Berikan jawaban yang informatif dan mudah dipahami.

Jenis pertanyaan: {{intent}}

user:
{{question}}

from promptflow.core import tool

@tool
def summarize(answer: str) -> str:
    """Ringkas dan format jawaban dari LLM."""
    if len(answer) > 500:
        return answer[:500] + "..."
    return answer

pf flow test --flow . --detail

pf flow test --flow . --inputs question="Bagaimana cara deploy model ML ke production?"

{"question": "Apa itu neural network?"}
{"question": "Bagaimana cara fine-tuning LLM?"}
{"question": "Bandingkan PyTorch dan TensorFlow"}
{"question": "Jelaskan konsep transfer learning"}

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

pf run show-details --name batchtestrun

Buat file chatflow.py:

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": (
                "Kamu adalah asisten AI yang membantu menjawab pertanyaan "
                "tentang data science dan machine learning. "
                "Jawab dengan bahasa yang mudah dipahami."
            )
        }

        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("Apa itu RAG dalam konteks LLM?")
    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:
    """Flow sederhana untuk question-answering."""
    from openai import OpenAI
    import os

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


    prompt = f"""Berdasarkan konteks berikut, jawab pertanyaan yang diberikan.

Konteks: {context if context else 'Tidak ada konteks khusus.'}

Pertanyaan: {question}

Jawaban:"""

    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
    }

Buat file evalflow.py:

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

@tool
def evaluateanswer(

    question: str,
    answer: str,
    groundtruth: str = ""
) -> dict:
    """Evaluasi kualitas jawaban LLM."""

    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:
    """Hitung skor relevansi jawaban terhadap pertanyaan."""
    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:

    """Hitung skor koherensi teks."""
    sentences = text.split(".")
    if len(sentences) <= 1:
        return 0.5
    return min(len(sentences) / 10, 1.0)


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

    """Hitung skor akurasi terhadap 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)

Menggunakan Built-in Evaluators

Prompt Flow menyediakan evaluator bawaan yang menggunakan LLM untuk menilai kualitas output:

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 untuk Debugging

Prompt Flow menyediakan tracing otomatis untuk memantau eksekusi flow:

from promptflow.tracing import trace, starttrace
import json

starttrace()

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

    """Proses dokumen dengan 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:
    """Pecah dokumen menjadi 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 untuk setiap 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 dari 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": "Buatkan ringkasan singkat dari teks berikut."},
            {"role": "user", "content": combined}
        ]
    )

    return response.choices[0].message.content

Jalankan trace viewer:

pf service start

Akses trace viewer di browser melalui http://localhost:23333.

Membangun RAG Flow

Salah satu use case paling umum adalah membangun 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 dokumen ke 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 dokumen relevan berdasarkan 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 jawaban berdasarkan context yang di-retrieve."""
        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": (
                        "Jawab pertanyaan berdasarkan konteks yang diberikan. "
                        "Jika jawaban tidak ada dalam konteks, katakan bahwa "
                        "informasi tidak tersedia."
                    )
                },
                {
                    "role": "user",
                    "content": f"Konteks:\n{contexttext}\n\nPertanyaan: {query}"
                }
            ],
            temperature=0.3
        )

        return response.choices[0].message.content

    @trace
    def call(self, query: str) -> dict:
        """Jalankan 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": "Apa itu 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. Manajemen Prompt

Pisahkan prompt dari kode logic untuk memudahkan iterasi:

from pathlib import Path
from promptflow.core import tool

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


@tool
def getprompt(intent: str, language: str = "id") -> str:
    """Load prompt template berdasarkan intent dan bahasa."""
    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:
    """Panggil LLM dengan 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("Bagaimana cara deploy?") == "tutorial"
    assert classifyintent("Apa itu ML?") == "explanation"

    assert classifyintent("Bandingkan A dan B") == "comparison"

def testflowoutput():
    from promptflow.client import PFClient

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

5. Monitoring dan Observability

Integrasikan dengan Application Insights untuk monitoring di production:

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)

Kesimpulan

Prompt Flow adalah framework yang sangat berguna untuk membangun aplikasi LLM yang terstruktur dan production-ready. Dengan pendekatan flow-based, developer dapat membangun pipeline yang jelas, mudah di-debug, dan dapat dievaluasi secara sistematis.

Keunggulan utama Prompt Flow meliputi kemudahan dalam mendefinisikan flow sebagai DAG, built-in evaluation framework, tracing yang komprehensif, dan integrasi yang baik dengan ekosistem Azure. Framework ini cocok untuk tim yang ingin membangun aplikasi LLM dengan standar engineering yang tinggi.

Untuk langkah selanjutnya, cobalah membangun flow yang lebih kompleks seperti multi-agent system atau RAG pipeline dengan evaluasi otomatis. Dokumentasi resmi Prompt Flow di GitHub dan Microsoft Learn menyediakan banyak contoh dan referensi tambahan.