DSPy: Framework untuk Optimasi LLM Secara Programatik

Prompt engineering secara manual adalah proses yang melelahkan dan sulit di-maintain. Setiap kali model berubah atau data berubah, prompt harus ditulis ulang. DSPy (Declarative Self-improving Python) hadir sebagai solusi revolusioner: framework yang memungkinkan Anda mengoptimasi prompt LLM secara programatik, layaknya melatih model machine learning.

Dalam tutorial ini, kita akan mempelajari cara menggunakan DSPy untuk membangun pipeline LLM yang dapat dioptimasi secara otomatis, mulai dari konsep dasar hingga implementasi RAG (Retrieval-Augmented Generation) yang lengkap.

Apa Itu DSPy?

DSPy adalah framework Python yang mengubah cara kita bekerja dengan LLM. Alih-alih menulis prompt secara manual, DSPy memungkinkan Anda mendefinisikan apa yang ingin dicapai (melalui signatures dan modules), lalu secara otomatis mengoptimasi bagaimana mencapainya (melalui optimizers/teleprompters).

Analogi sederhananya: jika prompt engineering manual seperti menulis kode assembly, DSPy seperti menggunakan compiler yang mengoptimasi kode Anda secara otomatis.

Konsep utama DSPy:

• Signatures: Mendefinisikan input dan output dari sebuah task
• Modules: Building blocks yang mengimplementasikan strategi prompting
• Optimizers (Teleprompters): Algoritma yang mengoptimasi prompt secara otomatis
• Metrics: Fungsi evaluasi untuk mengukur kualitas output
• Compilation: Proses menggabungkan semua komponen dan mengoptimasi

Instalasi

Install DSPy dan dependensi yang diperlukan:

pip install dspy-ai openai

Untuk fitur tambahan:

# Untuk retrieval dengan ChromaDB
pip install dspy-ai chromadb

Untuk menggunakan model lokal
pip install dspy-ai transformers torch

Konfigurasi API key:

export OPENAIAPIKEY="sk-your-api-key-here"

Setup dan Konfigurasi Dasar

Langkah pertama adalah mengkonfigurasi LLM yang akan digunakan.

import dspy

Konfigurasi dengan OpenAI
lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)

Atau dengan model lain
lm = dspy.LM("anthropic/claude-sonnet-4-20250514")
lm = dspy.LM("openai/gpt-4o")

Signatures: Mendefinisikan Task

Signatures adalah cara DSPy mendefinisikan input dan output dari sebuah task. Ini adalah abstraksi paling dasar di DSPy.

# Format: "input -> output"
Classify sentiment
classify = dspy.Predict("text -> sentiment")
result = classify(text="Produk ini sangat bagus dan berkualitas!")
print(result.sentiment)  # positive

Question answering
qa = dspy.Predict("question -> answer")
result = qa(question="Apa ibukota Indonesia?")
print(result.answer)  # Jakarta

Summarization
summarize = dspy.Predict("document -> summary")
result = summarize(
    document="Artikel panjang tentang AI dan dampaknya..."
)
print(result.summary)

class SentimentAnalysis(dspy.Signature):
    """Analisis sentimen dari teks review produk dalam bahasa Indonesia."""

    text: str = dspy.InputField(desc="Teks review produk")
    sentiment: str = dspy.OutputField(
        desc="Sentimen: positif, negatif, atau netral"
    )
    confidence: float = dspy.OutputField(
        desc="Tingkat keyakinan antara 0.0 dan 1.0"
    )
    reasoning: str = dspy.OutputField(
        desc="Alasan di balik klasifikasi sentimen"
    )

Gunakan signature
predictor = dspy.Predict(SentimentAnalysis)
result = predictor(
    text="Barang sampai tepat waktu, kualitas sesuai deskripsi. "
         "Tapi packaging kurang rapi."
)
print(f"Sentimen: {result.sentiment}")
print(f"Confidence: {result.confidence}")
print(f"Reasoning: {result.reasoning}")

Modules: Building Blocks

DSPy menyediakan beberapa module bawaan yang mengimplementasikan strategi prompting berbeda.

dspy.Predict

Module paling dasar yang melakukan single LLM call.

predictor = dspy.Predict("question -> answer")
result = predictor(question="Jelaskan konsep machine learning")
print(result.answer)

dspy.ChainOfThought

Menambahkan reasoning step sebelum menghasilkan output, mengimplementasikan teknik Chain-of-Thought prompting secara otomatis.

class MathProblem(dspy.Signature):
    """Selesaikan soal matematika langkah demi langkah."""
    problem: str = dspy.InputField(desc="Soal matematika")
    answer: float = dspy.OutputField(desc="Jawaban numerik")

Dengan ChainOfThought, model akan berpikir step-by-step
cot = dspy.ChainOfThought(MathProblem)
result = cot(
    problem="Sebuah toko menjual 150 barang dengan harga "
            "Rp 25.000 per barang. Diskon 15%. "
            "Berapa total pendapatan?"
)
print(f"Reasoning: {result.reasoning}")
print(f"Answer: {result.answer}")

dspy.ReAct

Mengimplementasikan pola Reasoning + Acting, di mana model bisa menggunakan tools untuk menyelesaikan task.

def searchdatabase(query: str) -> str:
    """Simulasi pencarian database."""
    data = {
        "python": "Python adalah bahasa pemrograman tingkat tinggi",
        "javascript": "JavaScript adalah bahasa pemrograman untuk web",
        "rust": "Rust adalah bahasa pemrograman sistem yang aman",
    }
    for key, value in data.items():
        if key in query.lower():
            return value
    return "Data tidak ditemukan"

def calculate(expression: str) -> str:
    """Hitung ekspresi matematika."""
    try:
        return str(eval(expression))
    except Exception:
        return "Error dalam kalkulasi"

ReAct dengan tools
react = dspy.ReAct(
    "question -> answer",
    tools=[searchdatabase, calculate],
)

result = react(
    question="Apa itu Python dan berapa 15  23?"

)
print(result.answer)

cothint = dspy.ChainOfThoughtWithHint("question -> answer")
result = cothint(
    question="Bagaimana cara mengoptimasi query SQL yang lambat?",
    hint="Pertimbangkan indexing, query plan, dan normalisasi data"
)
print(result.answer)

class ArticleGenerator(dspy.Module):
    def init(self):
        super().init()
        self.outline = dspy.ChainOfThought(
            "topic, audience -> outline"
        )
        self.draft = dspy.ChainOfThought(
            "topic, outline -> article"
        )
        self.review = dspy.Predict(
            "article -> improvedarticle, feedback"

        )

    def forward(self, topic: str, audience: str):
        # Step 1: Buat outline
        outlineresult = self.outline(
            topic=topic, audience=audience
        )

        # Step 2: Tulis draft berdasarkan outline
        draftresult = self.draft(

            topic=topic,
            outline=outlineresult.outline
        )

        # Step 3: Review dan perbaiki
        finalresult = self.review(

            article=draftresult.article
        )

        return dspy.Prediction(
            outline=outlineresult.outline,

            article=finalresult.improvedarticle,

            feedback=finalresult.feedback,
        )

Gunakan module
generator = ArticleGenerator()
result = generator(
    topic="Penerapan AI dalam Pertanian Indonesia",
    audience="Petani dan pelaku agrikultur"
)
print(f"Outline: {result.outline}")
print(f"Article: {result.article}")

# Metric sederhana: exact match
def exactmatch(example, prediction, trace=None):

    return example.answer.lower() == prediction.answer.lower()

Metric dengan scoring partial
def qualitymetric(example, prediction, trace=None):
    score = 0.0

    # Cek apakah jawaban mengandung kata kunci penting
    keywords = example.get("keywords", [])
    if keywords:
        matches = sum(
            1 for kw in keywords
            if kw.lower() in prediction.answer.lower()
        )
        score += matches / len(keywords)  0.5

    # Cek panjang jawaban (tidak terlalu pendek/panjang)
    wordcount = len(prediction.answer.split())

    if 50 <= wordcount <= 300:
        score += 0.3
    elif 20 <= wordcount < 50:

        score += 0.15

    # Cek apakah ada reasoning
    if hasattr(prediction, "reasoning") and prediction.reasoning:
        score += 0.2

    return score

Metric menggunakan LLM sebagai judge
class AnswerQuality(dspy.Signature):
    """Evaluasi kualitas jawaban berdasarkan pertanyaan dan referensi."""
    question: str = dspy.InputField()
    referenceanswer: str = dspy.InputField()
    predictedanswer: str = dspy.InputField()

    qualityscore: float = dspy.OutputField(
        desc="Skor kualitas antara 0.0 dan 1.0"
    )

def llmjudgemetric(example, prediction, trace=None):
    judge = dspy.Predict(AnswerQuality)
    result = judge(
        question=example.question,
        referenceanswer=example.answer,

        predictedanswer=prediction.answer,
    )
    return float(result.qualityscore)

# Buat dataset training
trainset = [
    dspy.Example(
        question="Apa itu machine learning?",
        answer="Machine learning adalah cabang kecerdasan buatan "
               "yang memungkinkan komputer belajar dari data "
               "tanpa diprogram secara eksplisit."
    ).withinputs("question"),
    dspy.Example(
        question="Jelaskan perbedaan supervised dan unsupervised learning",
        answer="Supervised learning menggunakan data berlabel untuk "
               "training, sedangkan unsupervised learning menemukan "
               "pola dari data tanpa label."
    ).withinputs("question"),

    dspy.Example(
        question="Apa fungsi activation function dalam neural network?",
        answer="Activation function menentukan output neuron "
               "berdasarkan input yang diterima, menambahkan "
               "non-linearitas ke dalam model."
    ).withinputs("question"),
    # Tambahkan lebih banyak contoh...
]

devset = [
    dspy.Example(
        question="Apa itu transfer learning?",
        answer="Transfer learning adalah teknik menggunakan model "
               "yang sudah dilatih pada task tertentu sebagai "
               "titik awal untuk task baru yang berbeda."
    ).withinputs("question"),

]

from dspy.teleprompt import BootstrapFewShot

Definisikan module
qamodule = dspy.ChainOfThought("question -> answer")

Definisikan metric
def answerquality(example, prediction, trace=None):

    # Sederhana: cek apakah prediksi cukup panjang dan relevan
    if len(prediction.answer.split()) < 10:
        return False
    return True

Compile dengan BootstrapFewShot
optimizer = BootstrapFewShot(
    metric=answerquality,
    maxbootstrappeddemos=4,
    maxlabeleddemos=4,
)

compiledqa = optimizer.compile(

    qamodule,
    trainset=trainset,
)

Gunakan module yang sudah dioptimasi
result = compiledqa(question="Apa itu deep learning?")

print(result.answer)

from dspy.teleprompt import BootstrapFewShotWithRandomSearch

optimizer = BootstrapFewShotWithRandomSearch(
    metric=answerquality,
    maxbootstrappeddemos=4,
    maxlabeleddemos=4,
    numcandidateprograms=10,
    numthreads=4,

)

compiledqa = optimizer.compile(
    qamodule,

    trainset=trainset,
    valset=devset,
)

from dspy.teleprompt import MIPROv2

optimizer = MIPROv2(
    metric=answerquality,
    numcandidates=7,

    inittemperature=1.0,
)

compiledqa = optimizer.compile(

    qamodule,
    trainset=trainset,
    maxbootstrappeddemos=4,
    maxlabeleddemos=4,
)

Saving dan Loading Compiled Programs

Setelah optimasi, Anda bisa menyimpan dan memuat ulang program yang sudah dicompile.

# Simpan program yang sudah dioptimasi
compiledqa.save("optimizedqaprogram.json")


Muat kembali
loadedqa = dspy.ChainOfThought("question -> answer")
loadedqa.load("optimizedqaprogram.json")


Gunakan seperti biasa
result = loadedqa(question="Apa itu neural network?")
print(result.answer)

Contoh Praktis: RAG Pipeline dengan DSPy

Berikut implementasi lengkap RAG (Retrieval-Augmented Generation) menggunakan DSPy.

import dspy
from dspy.retrieve.chromadbrm import ChromadbRM

import chromadb

Setup ChromaDB
chromaclient = chromadb.PersistentClient(path="./chromadb")


Buat collection dan tambahkan dokumen
collection = chromaclient.getorcreatecollection("knowledgebase")

Tambahkan dokumen
documents = [
    "Python adalah bahasa pemrograman yang dibuat oleh Guido van Rossum "
    "pada tahun 1991. Python terkenal karena sintaksnya yang mudah "
    "dibaca dan dipelajari.",
    "Machine learning memiliki tiga paradigma utama: supervised learning, "
    "unsupervised learning, dan reinforcement learning. Masing-masing "
    "memiliki use case yang berbeda.",
    "Deep learning adalah subset dari machine learning yang menggunakan "
    "neural network dengan banyak layer. Framework populer termasuk "
    "TensorFlow, PyTorch, dan JAX.",
    "Natural Language Processing (NLP) adalah cabang AI yang berfokus "
    "pada interaksi antara komputer dan bahasa manusia. Tugas NLP "
    "meliputi sentiment analysis, named entity recognition, "
    "dan machine translation.",
    "MLOps adalah praktik yang menggabungkan Machine Learning dan "
    "DevOps untuk mengotomasi dan menstandarkan lifecycle model ML "
    "dari development hingga production.",
]

collection.add(
    documents=documents,
    ids=[f"doc{i}" for i in range(len(documents))],

)

Setup retriever
retriever = ChromadbRM(
    collectionname="knowledgebase",

    persistdirectory="./chromadb",

    k=3,
)

dspy.configure(
    lm=dspy.LM("openai/gpt-4o-mini"),
    rm=retriever,
)

Definisikan RAG Signature
class GenerateAnswer(dspy.Signature):
    """Jawab pertanyaan berdasarkan konteks yang diberikan.
    Berikan jawaban yang informatif dan akurat."""

    context: list[str] = dspy.InputField(
        desc="Konteks relevan dari knowledge base"
    )
    question: str = dspy.InputField(
        desc="Pertanyaan yang harus dijawab"
    )
    answer: str = dspy.OutputField(
        desc="Jawaban yang detail dan akurat berdasarkan konteks"
    )

Bangun RAG Module
class RAGModule(dspy.Module):
    def init(self, numpassages=3):
        super().init()
        self.retrieve = dspy.Retrieve(k=numpassages)

        self.generate = dspy.ChainOfThought(GenerateAnswer)

    def forward(self, question):
        # Retrieve relevant passages
        context = self.retrieve(question).passages

        # Generate answer
        result = self.generate(
            context=context,
            question=question,
        )

        return dspy.Prediction(
            context=context,
            answer=result.answer,
            reasoning=result.reasoning,
        )

Buat instance dan gunakan
rag = RAGModule()
result = rag(question="Apa saja paradigma utama dalam machine learning?")
print(f"Answer: {result.answer}")
print(f"Reasoning: {result.reasoning}")
print(f"Context used: {result.context}")

Optimasi RAG dengan data training
trainsetrag = [
    dspy.Example(
        question="Siapa pembuat Python?",
        answer="Python dibuat oleh Guido van Rossum pada tahun 1991."
    ).withinputs("question"),

    dspy.Example(
        question="Apa itu deep learning?",
        answer="Deep learning adalah subset dari machine learning "
               "yang menggunakan neural network dengan banyak layer."
    ).withinputs("question"),
    dspy.Example(
        question="Sebutkan framework deep learning yang populer",
        answer="Framework deep learning populer termasuk TensorFlow, "
               "PyTorch, dan JAX."
    ).withinputs("question"),

]

Compile RAG module
from dspy.teleprompt import BootstrapFewShot

def ragmetric(example, prediction, trace=None):
    # Cek apakah jawaban mengandung informasi dari referensi
    refwords = set(example.answer.lower().split())

    predwords = set(prediction.answer.lower().split())
    overlap = len(refwords & predwords) / len(refwords)

    return overlap > 0.5

optimizer = BootstrapFewShot(
    metric=ragmetric,
    maxbootstrappeddemos=2,
)

compiledrag = optimizer.compile(rag, trainset=trainsetrag)

Gunakan RAG yang sudah dioptimasi
result = compiledrag(

    question="Apa saja tugas-tugas dalam NLP?"
)
print(f"Optimized Answer: {result.answer}")

from dspy.evaluate import Evaluate

Buat evaluator
evaluator = Evaluate(
    devset=devset,
    metric=ragmetric,
    numthreads=4,

    displayprogress=True,
    displaytable=5,

)

Evaluasi module sebelum optimasi
scorebefore = evaluator(rag)
print(f"Score sebelum optimasi: {scorebefore}")


Evaluasi module setelah optimasi
scoreafter = evaluator(compiledrag)

print(f"Score setelah optimasi: {scoreafter}")

Tips dan Best Practices

Deskripsi yang baik dalam signature membantu LLM dan optimizer memahami task dengan lebih baik.

# Baik
class GoodSignature(dspy.Signature):
    """Ekstrak entitas bisnis dari laporan keuangan perusahaan Indonesia."""
    report: str = dspy.InputField(desc="Teks laporan keuangan")
    entities: list[str] = dspy.OutputField(
        desc="Daftar nama perusahaan, angka keuangan, dan metrik utama"
    )

Kualitas data training sangat mempengaruhi hasil optimasi. Pastikan contoh-contoh Anda representatif dan akurat.

• BootstrapFewShot: Cepat, bagus untuk memulai
• BootstrapFewShotWithRandomSearch: Lebih baik tapi lebih lambat
• MIPROv2: Terbaik untuk production, tapi paling lambat

Selalu evaluasi sebelum dan sesudah optimasi. Jika hasilnya tidak memuaskan, coba tambah data training atau gunakan optimizer berbeda.

Setelah menemukan konfigurasi optimal, selalu simpan hasilnya agar tidak perlu mengulang proses optimasi.

Kesimpulan

DSPy mengubah paradigma dalam bekerja dengan LLM. Alih-alih menghabiskan waktu menulis dan memperbaiki prompt secara manual, DSPy memungkinkan Anda fokus pada mendefinisikan task dan membiarkan framework mengoptimasi implementasinya.

Poin-poin utama yang telah kita pelajari:

• Signatures untuk mendefinisikan input/output task
• Modules seperti ChainOfThought dan ReAct untuk strategi prompting
• Optimizers untuk mengoptimasi prompt secara otomatis
• Metrics untuk mengukur kualitas output
• Implementasi RAG pipeline yang lengkap
• Evaluasi dan saving/loading compiled programs

Dengan DSPy, Anda bisa membangun pipeline LLM yang lebih robust, maintainable, dan performant. Mulailah dengan task sederhana, optimasi secara bertahap, dan skalakan sesuai kebutuhan proyek Anda.