Tutorial Marvin AI: Toolkit AI Engineering yang Ringan dan Powerful

Dalam tutorial ini, kita akan mempelajari semua fitur utama Marvin mulai dari klasifikasi teks hingga membangun data transformation pipeline yang kompleks.

Apa itu Marvin?

Marvin adalah toolkit AI engineering yang dikembangkan oleh tim Prefect. Library ini dirancang dengan filosofi bahwa AI seharusnya menjadi komponen software biasa - bukan sesuatu yang memerlukan setup rumit atau abstraksi berlapis-lapis.

Fitur utama Marvin:

• aifn: Ubah function Python biasa menjadi AI-powered function
• aiclassifier: Klasifikasi teks ke dalam kategori yang telah ditentukan
• aiextract: Ekstraksi entity dan informasi terstruktur dari teks
• aimodel: Generate structured data menggunakan Pydantic models
• Image processing: Analisis dan pemrosesan gambar dengan AI
• Casting/Mapping: Transformasi data dengan AI
• Async support: Dukungan penuh untuk operasi asynchronous

Instalasi dan Setup

Instalasi Package

# Instalasi Marvin
pip install marvin

Untuk fitur image processing
pip install "marvin[image]"

Untuk semua fitur
pip install "marvin[all]"

Konfigurasi API Key

# Set environment variable untuk OpenAI
export OPENAIAPIKEY="sk-your-api-key-here"

# Atau konfigurasi via Python
import marvin

marvin.settings.openai.apikey = "sk-your-api-key-here"


Opsional: set model default
marvin.settings.openai.chat.completions.model = "gpt-4o-mini"

AI Functions (aifn)

import marvin

@marvin.fn
def sentiment(text: str) -> str:
    """Analisis sentimen teks dan kembalikan 'positif', 'negatif', atau 'netral'."""

result = sentiment("Produk ini luar biasa! Sangat memuaskan!")
print(result)  # Output: positif

result = sentiment("Pelayanannya sangat mengecewakan")
print(result)  # Output: negatif

from typing import List, Dict
import marvin

@marvin.fn
def generatehashtags(topic: str, count: int = 5) -> List[str]:
    """Generate hashtag yang relevan untuk topik yang diberikan.
    Hashtag harus populer dan relevan untuk social media."""

hashtags = generatehashtags("machine learning Indonesia", count=7)

print(hashtags)
Output: ['#MachineLearning', '#AIIndonesia', '#DataScience', ...]

@marvin.fn
def translatephrases(phrases: List[str], targetlanguage: str) -> Dict[str, str]:

    """Terjemahkan setiap frasa ke bahasa target.
    Kembalikan dictionary dengan frasa asli sebagai key dan terjemahan sebagai value."""

result = translatephrases(
    ["Good morning", "Thank you", "How are you?"],
    targetlanguage="Indonesian"

)
print(result)
Output: {'Good morning': 'Selamat pagi', 'Thank you': 'Terima kasih', ...}

import marvin

@marvin.fn
def categorizeexpense(description: str) -> str:
    """Kategorikan pengeluaran bisnis ke salah satu kategori:
    'travel', 'meals', 'officesupplies', 'software', 'marketing',

    'utilities', 'payroll', 'other'."""

@marvin.fn
def extractactionitems(meetingnotes: str) -> List[str]:
    """Ekstrak action items dari catatan rapat.
    Setiap item harus jelas dan actionable."""

@marvin.fn
def generateemailreply(
    originalemail: str,

    tone: str = "professional",
    keypoints: List[str] = None
) -> str:
    """Buat balasan email berdasarkan email asli.
    Gunakan tone yang ditentukan dan sertakan key points jika ada."""

Penggunaan
category = categorizeexpense("Uber ride to client meeting")

print(category)  # Output: travel

actions = extractactionitems("""

Rapat Tim Data Science - 28 April 2026
Budi akan membuat prototype model NLP minggu depan
Sari perlu review dataset baru sebelum Jumat
Tim perlu setup monitoring untuk model production
""")

print(actions)

AI Classifier (aiclassifier)

import marvin
from enum import Enum

class Sentiment(Enum):
    POSITIVE = "positive"
    NEGATIVE = "negative"
    NEUTRAL = "neutral"

result = marvin.classify("Aplikasi ini sangat membantu pekerjaan saya!", Sentiment)
print(result)  # Output: Sentiment.POSITIVE

Klasifikasi batch
texts = [
    "Produk bagus sekali!",
    "Biasa saja, tidak ada yang istimewa",
    "Sangat kecewa dengan layanan ini",
    "Harganya sesuai dengan kualitas"
]

results = [marvin.classify(text, Sentiment) for text in texts]
for text, result in zip(texts, results):
    print(f"{text} -> {result.value}")

from enum import Enum
import marvin

class ContentCategory(Enum):
    SAFE = "safe"
    SPAM = "spam"
    OFFENSIVE = "offensive"
    MISINFORMATION = "misinformation"
    ADULT = "adult"

class UrgencyLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

Content moderation
def moderatecontent(text: str) -> dict:
    category = marvin.classify(text, ContentCategory)
    urgency = marvin.classify(text, UrgencyLevel)

    return {
        "text": text,
        "category": category.value,
        "urgency": urgency.value,
        "issafe": category == ContentCategory.SAFE

    }

Test
result = moderatecontent("Beli produk kami sekarang! Klik link ini untuk diskon 90%!")
print(result)
Output: {'text': '...', 'category': 'spam', 'urgency': 'medium', 'issafe': False}

AI Extract (aiextract)

import marvin

Ekstraksi sederhana
names = marvin.extract("Budi dan Sari pergi ke kantor bersama Andi", str, 
                        instructions="extract person names")
print(names)  # Output: ['Budi', 'Sari', 'Andi']

Ekstraksi angka
numbers = marvin.extract(
    "Harga laptop Rp15.000.000 dan mouse Rp250.000", 
    float,
    instructions="extract prices as numbers"
)
print(numbers)  # Output: [15000000.0, 250000.0]

from pydantic import BaseModel, Field
from typing import Optional, List
import marvin

class ContactInfo(BaseModel):
    name: str = Field(description="Nama lengkap")
    email: Optional[str] = Field(description="Alamat email")
    phone: Optional[str] = Field(description="Nomor telepon")
    company: Optional[str] = Field(description="Nama perusahaan")

Ekstrak kontak dari teks
text = """
Silakan hubungi Budi Santoso di budi.santoso@techcorp.id atau 
telepon 081234567890. Untuk urusan marketing, kontak Sari Dewi 
dari PT Digital Nusantara di sari@digitalnusantara.co.id.
"""

contacts = marvin.extract(text, ContactInfo)
for contact in contacts:
    print(f"Nama: {contact.name}")
    print(f"Email: {contact.email}")
    print(f"Phone: {contact.phone}")
    print(f"Company: {contact.company}")
    print()

class InvoiceItem(BaseModel):
    description: str
    quantity: int
    unitprice: float
    total: float

class InvoiceData(BaseModel):
    invoicenumber: str

    date: str
    vendor: str
    items: List[InvoiceItem]
    subtotal: float
    tax: float
    total: float

invoicetext = """
Invoice #INV-2026-0042
Tanggal: 15 April 2026
Dari: PT Teknologi Maju
Laptop Dell XPS 15 - 2 unit @ Rp 22.000.000 = Rp 44.000.000
Monitor LG 27" 4K - 2 unit @ Rp 6.500.000 = Rp 13.000.000
Keyboard Mechanical - 3 unit @ Rp 1.200.000 = Rp 3.600.000

Subtotal: Rp 60.600.000
PPN 11%: Rp 6.666.000
Total: Rp 67.266.000
"""

invoicedata = marvin.extract(invoicetext, InvoiceData)

AI Model (aimodel)

import marvin
from pydantic import BaseModel, Field
from typing import List, Optional

class Product(BaseModel):
    name: str = Field(description="Nama produk")
    description: str = Field(description="Deskripsi produk")
    pricerange: str = Field(description="Range harga")

    targetaudience: str = Field(description="Target konsumen")
    features: List[str] = Field(description="Fitur utama")

Generate product data
product = marvin.model(
    Product,
    instructions="Buat deskripsi untuk aplikasi mobile fitness tracker Indonesia"
)
print(f"Produk: {product.name}")
print(f"Deskripsi: {product.description}")
print(f"Harga: {product.pricerange}")

print(f"Target: {product.targetaudience}")
print(f"Fitur: {', '.join(product.features)}")

Generate Multiple Models

class UserPersona(BaseModel):
    name: str
    age: int
    occupation: str
    goals: List[str]
    painpoints: List[str]

    techsavviness: str

Generate beberapa persona
personas = [
    marvin.model(
        UserPersona,
        instructions=f"Buat user persona #{i+1} untuk aplikasi e-learning Indonesia"
    )
    for i in range(3)
]

for persona in personas:
    print(f"\n{'='40}")

    print(f"Nama: {persona.name}, Usia: {persona.age}")
    print(f"Pekerjaan: {persona.occupation}")
    print(f"Goals: {', '.join(persona.goals)}")
    print(f"Pain Points: {', '.join(persona.painpoints)}")

import marvin
from pathlib import Path

Deskripsi gambar
description = marvin.describeimage(
    Path("./productphoto.jpg"),

    instructions="Deskripsikan produk dalam gambar ini secara detail"
)
print(description)

Klasifikasi gambar
from enum import Enum

class ImageCategory(Enum):
    PRODUCT = "product"
    LANDSCAPE = "landscape"
    PEOPLE = "people"
    FOOD = "food"
    DOCUMENT = "document"

category = marvin.classifyimage(
    Path("./photo.jpg"),
    ImageCategory
)
print(f"Kategori: {category.value}")

from pydantic import BaseModel
from typing import List
import marvin

class ReceiptItem(BaseModel):
    name: str
    price: float

class ReceiptData(BaseModel):
    storename: str

    items: List[ReceiptItem]
    total: float

Ekstrak data dari foto struk
receipt = marvin.extractimage(
    Path("./receipt.jpg"),
    ReceiptData,
    instructions="Ekstrak informasi dari struk belanja ini"
)
print(f"Toko: {receipt.storename}")

print(f"Total: Rp {receipt.total:,.0f}")

import marvin

Cast string ke format yang diinginkan
result = marvin.cast("dua puluh lima ribu", int)
print(result)  # Output: 25000

result = marvin.cast("15 April 2026", str, 
                      instructions="convert to ISO format date")
print(result)  # Output: 2026-04-15

Cast ke Pydantic model
from pydantic import BaseModel

class Address(BaseModel):
    street: str
    city: str
    province: str
    postalcode: str

address = marvin.cast(
    "Jl. Sudirman No. 123, Menteng, Jakarta Pusat 10310",
    Address
)
print(f"Jalan: {address.street}")
print(f"Kota: {address.city}")
print(f"Kode Pos: {address.postalcode}")

import marvin

Map untuk batch transformation
fruits = ["apel", "jeruk", "mangga", "durian", "rambutan"]
translations = marvin.map(fruits, str, instructions="translate to English")
print(dict(zip(fruits, translations)))
Output: {'apel': 'apple', 'jeruk': 'orange', 'mangga': 'mango', ...}

Map dengan tipe output kompleks
descriptions = [
    "Software engineer, 5 tahun pengalaman, Jakarta",
    "Data scientist, fresh graduate, Bandung",
    "Product manager, 3 tahun, remote"
]

class CandidateProfile(BaseModel):
    role: str
    experienceyears: int
    location: str
    seniority: str

profiles = marvin.map(descriptions, CandidateProfile)
for desc, profile in zip(descriptions, profiles):
    print(f"{profile.role} - {profile.seniority} ({profile.location})")

import asyncio
import marvin

Async AI function
@marvin.fn
async def asyncsummarize(text: str) -> str:

    """Buat ringkasan singkat dari teks yang diberikan dalam 2-3 kalimat."""

Async classification
async def processbatch(texts: List[str]):
    tasks = [
        marvin.classifyasync(text, Sentiment)

        for text in texts
    ]
    results = await asyncio.gather(tasks)
    return results

Async extraction
async def extractallcontacts(documents: List[str]):

    tasks = [
        marvin.extractasync(doc, ContactInfo)
        for doc in documents
    ]
    results = await asyncio.gather(tasks)

    return results

Jalankan async
async def main():
    texts = [
        "Produk sangat bagus!",
        "Layanan mengecewakan",
        "Biasa saja",
        "Luar biasa! Pasti beli lagi",
        "Tidak sesuai ekspektasi"
    ]

    results = await processbatch(texts)

    for text, sentiment in zip(texts, results):
        print(f"{text} -> {sentiment.value}")

asyncio.run(main())

import marvin
from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum
from datetime import datetime

Step 1: Definisikan schema dan klasifikasi
class FeedbackCategory(Enum):
    PRODUCT = "product"
    SERVICE = "service"
    PRICING = "pricing"
    UX = "userexperience"
    DELIVERY = "delivery"
    OTHER = "other"

class SentimentLevel(Enum):
    VERYPOSITIVE = "verypositive"
    POSITIVE = "positive"
    NEUTRAL = "neutral"
    NEGATIVE = "negative"
    VERYNEGATIVE = "verynegative"

class FeedbackEntity(BaseModel):
    productmentioned: Optional[str] = None

    featurementioned: Optional[str] = None
    competitormentioned: Optional[str] = None

    issuedescribed: Optional[str] = None

class ProcessedFeedback(BaseModel):
    originaltext: str

    category: str
    sentiment: str
    entities: FeedbackEntity
    summary: str
    priorityscore: int = Field(description="1-10, 10 being highest priority")
    suggestedaction: str


Step 2: Buat AI functions
@marvin.fn
def generatesummary(feedback: str) -> str:
    """Buat ringkasan feedback pelanggan dalam satu kalimat singkat."""

@marvin.fn
def suggestaction(feedback: str, category: str, sentiment: str) -> str:

    """Sarankan tindakan yang harus diambil berdasarkan feedback pelanggan.
    Pertimbangkan kategori dan sentimen untuk memberikan saran yang tepat."""

@marvin.fn
def calculatepriority(feedback: str, sentiment: str) -> int:
    """Hitung skor prioritas 1-10 berdasarkan feedback dan sentimen.
    10 = sangat urgent, 1 = tidak urgent.
    Pertimbangkan: severity masalah, potensi churn, dampak bisnis."""

Step 3: Pipeline utama
def processfeedback(text: str) -> ProcessedFeedback:

    # Klasifikasi
    category = marvin.classify(text, FeedbackCategory)
    sentiment = marvin.classify(text, SentimentLevel)

    # Ekstraksi entity
    entities = marvin.extract(text, FeedbackEntity)
    entity = entities[0] if entities else FeedbackEntity()

    # Generate summary dan action
    summary = generatesummary(text)
    action = suggestaction(text, category.value, sentiment.value)

    priority = calculatepriority(text, sentiment.value)

    return ProcessedFeedback(
        originaltext=text,

        category=category.value,
        sentiment=sentiment.value,
        entities=entity,
        summary=summary,
        priorityscore=priority,
        suggestedaction=action

    )

Step 4: Batch processing
def analyzefeedbackbatch(feedbacks: List[str]) -> dict:

    results = [processfeedback(fb) for fb in feedbacks]

    # Analisis agregat
    sentimentdist = {}

    categorydist = {}
    highpriority = []


    for r in results:
        sentimentdist[r.sentiment] = sentimentdist.get(r.sentiment, 0) + 1

        categorydist[r.category] = categorydist.get(r.category, 0) + 1

        if r.priorityscore >= 7:
            highpriority.append(r)


    return {
        "totalfeedbacks": len(results),
        "sentimentdistribution": sentimentdist,
        "categorydistribution": categorydist,
        "highprioritycount": len(highpriority),

        "highpriorityitems": highpriority,
        "allresults": results

    }

Step 5: Jalankan pipeline
feedbacks = [
    "Aplikasinya sering crash saat checkout. Sudah 3x kejadian minggu ini!",
    "Fitur baru recommendation engine-nya bagus banget, sangat membantu!",
    "Harga langganan naik tapi fitur tidak ada yang baru. Mending pindah ke kompetitor X.",
    "Pengiriman lambat, sudah 5 hari belum sampai. Customer service tidak responsif.",
    "UI redesign yang baru lebih clean dan mudah digunakan. Good job!",
    "Biasa aja sih, tidak ada yang spesial dibanding aplikasi lain.",
]

Proses semua feedback
analysis = analyzefeedbackbatch(feedbacks)


Tampilkan hasil
print(f"\n{'='60}")
print(f"LAPORAN ANALISIS FEEDBACK PELANGGAN")
print(f"{'='*60}")
print(f"Total Feedback: {analysis['totalfeedbacks']}")
print(f"\nDistribusi Sentimen:")
for sent, count in analysis['sentimentdistribution'].items():

    print(f"  {sent}: {count}")

print(f"\nDistribusi Kategori:")
for cat, count in analysis['categorydistribution'].items():
    print(f"  {cat}: {count}")

print(f"\nItem Prioritas Tinggi: {analysis['highprioritycount']}")
for item in analysis['highpriorityitems']:
    print(f"\n  [{item.priorityscore}/10] {item.summary}")

    print(f"  Kategori: {item.category} | Sentimen: {item.sentiment}")
    print(f"  Aksi: {item.suggestedaction}")

Tips dan Best Practices

from functools import lrucache

@lrucache(maxsize=100)
def cached_classify(text: str) -> str:
    return marvin.classify(text, Sentiment).value

try:
    result = marvin.classify(text, Sentiment)
except Exception as e:
    print(f"Classification failed: {e}")
    result = Sentiment.NEUTRAL  # fallback

Kesimpulan

Marvin AI adalah toolkit yang sempurna untuk developer yang ingin mengintegrasikan AI ke dalam aplikasi mereka tanpa kompleksitas framework besar. Dengan pendekatan "AI as a function", Marvin membuat AI engineering terasa natural seperti menulis kode Python biasa.

Keunggulan utama Marvin dibanding framework lain adalah kesederhanaannya. Anda tidak perlu memahami konsep chain, graph, atau agent. Cukup definisikan apa yang Anda inginkan melalui type hints dan docstring, dan Marvin akan menangani sisanya. Ini menjadikan Marvin pilihan ideal untuk rapid prototyping, microservices, dan integrasi AI ke aplikasi yang sudah ada.

Mulailah dengan fitur sederhana seperti classify atau fn, lalu kembangkan ke pipeline yang lebih kompleks sesuai kebutuhan. Selamat mencoba!