Marvin AI Tutorial: A Lightweight and Powerful AI Engineering Toolkit

In this tutorial, we will explore all of Marvin's key features from text classification to building complex data transformation pipelines.

What is Marvin?

Marvin is an AI engineering toolkit developed by the Prefect team. The library is built on the philosophy that AI should be a regular software component - not something that requires complex setup or layers of abstraction.

Key features of Marvin:

• aifn: Turn regular Python functions into AI-powered functions
• aiclassifier: Classify text into predefined categories
• aiextract: Extract entities and structured information from text
• aimodel: Generate structured data using Pydantic models
• Image processing: Analyze and process images with AI
• Casting/Mapping: Transform data with AI
• Async support: Full support for asynchronous operations

Installation and Setup

Package Installation

# Install Marvin
pip install marvin

For image processing features
pip install "marvin[image]"

For all features
pip install "marvin[all]"

API Key Configuration

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

# Or configure via Python
import marvin

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


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

AI Functions (aifn)

import marvin

@marvin.fn
def sentiment(text: str) -> str:
    """Analyze the sentiment of the text and return 'positive', 'negative', or 'neutral'."""

result = sentiment("This product is amazing! Highly satisfied!")
print(result)  # Output: positive

result = sentiment("The service was very disappointing")
print(result)  # Output: negative

from typing import List, Dict
import marvin

@marvin.fn
def generatehashtags(topic: str, count: int = 5) -> List[str]:
    """Generate relevant hashtags for the given topic.
    Hashtags should be popular and relevant for social media."""

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

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

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

    """Translate each phrase to the target language.
    Return a dictionary with original phrase as key and translation as value."""

result = translatephrases(
    ["Selamat pagi", "Terima kasih", "Apa kabar?"],
    targetlanguage="English"

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

import marvin

@marvin.fn
def categorizeexpense(description: str) -> str:
    """Categorize a business expense into one of these categories:
    'travel', 'meals', 'officesupplies', 'software', 'marketing',

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

@marvin.fn
def extractactionitems(meetingnotes: str) -> List[str]:
    """Extract action items from meeting notes.
    Each item should be clear and actionable."""

@marvin.fn
def generateemailreply(
    originalemail: str,

    tone: str = "professional",
    keypoints: List[str] = None
) -> str:
    """Generate an email reply based on the original email.
    Use the specified tone and include key points if provided."""

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

print(category)  # Output: travel

actions = extractactionitems("""

Data Science Team Meeting - April 28, 2026
Alex will create the NLP model prototype next week
Sarah needs to review the new dataset before Friday
Team needs to set up monitoring for production models
""")

print(actions)

AI Classifier (aiclassifier)

import marvin
from enum import Enum

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

result = marvin.classify("This app really helps with my work!", Sentiment)
print(result)  # Output: Sentiment.POSITIVE

Batch classification
texts = [
    "Excellent product quality!",
    "Nothing special about it",
    "Very disappointed with the service",
    "Price matches the quality"
]

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("Buy our product now! Click this link for 90% discount!")
print(result)
Output: {'text': '...', 'category': 'spam', 'urgency': 'medium', 'issafe': False}

AI Extract (aiextract)

import marvin

Simple extraction
names = marvin.extract("Alex and Sarah went to the office with Jordan", str,
                        instructions="extract person names")
print(names)  # Output: ['Alex', 'Sarah', 'Jordan']

Number extraction
numbers = marvin.extract(
    "The laptop costs $1,500 and the mouse costs $25",
    float,
    instructions="extract prices as numbers"
)
print(numbers)  # Output: [1500.0, 25.0]

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

class ContactInfo(BaseModel):
    name: str = Field(description="Full name")
    email: Optional[str] = Field(description="Email address")
    phone: Optional[str] = Field(description="Phone number")
    company: Optional[str] = Field(description="Company name")

Extract contacts from text
text = """
Please contact Alex Johnson at alex.johnson@techcorp.com or 
call +1-555-0123. For marketing inquiries, reach out to Sarah Chen 
from Digital Innovations Inc at sarah@digitalinnovations.com.
"""

contacts = marvin.extract(text, ContactInfo)
for contact in contacts:
    print(f"Name: {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
Date: April 15, 2026
From: TechSupply Corp
Dell XPS 15 Laptop - 2 units @ $1,800 = $3,600
LG 27" 4K Monitor - 2 units @ $450 = $900
Mechanical Keyboard - 3 units @ $120 = $360

Subtotal: $4,860
Tax (10%): $486
Total: $5,346
"""

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="Product name")
    description: str = Field(description="Product description")
    pricerange: str = Field(description="Price range")

    targetaudience: str = Field(description="Target consumers")
    features: List[str] = Field(description="Key features")

Generate product data
product = marvin.model(
    Product,
    instructions="Create a description for a mobile fitness tracker app"
)
print(f"Product: {product.name}")
print(f"Description: {product.description}")
print(f"Price: {product.pricerange}")

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

Generate Multiple Models

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

    techsavviness: str

Generate multiple personas
personas = [
    marvin.model(
        UserPersona,
        instructions=f"Create user persona #{i+1} for an e-learning platform"
    )
    for i in range(3)
]

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

    print(f"Name: {persona.name}, Age: {persona.age}")
    print(f"Occupation: {persona.occupation}")
    print(f"Goals: {', '.join(persona.goals)}")
    print(f"Pain Points: {', '.join(persona.painpoints)}")

import marvin
from pathlib import Path

Describe an image
description = marvin.describeimage(
    Path("./productphoto.jpg"),

    instructions="Describe the product in this image in detail"
)
print(description)

Classify an image
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"Category: {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

Extract data from receipt photo
receipt = marvin.extractimage(
    Path("./receipt.jpg"),
    ReceiptData,
    instructions="Extract information from this shopping receipt"
)
print(f"Store: {receipt.storename}")

print(f"Total: ${receipt.total:.2f}")

import marvin

Cast string to desired format
result = marvin.cast("twenty-five thousand", int)
print(result)  # Output: 25000

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

Cast to Pydantic model
from pydantic import BaseModel

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

address = marvin.cast(
    "123 Main Street, Suite 456, San Francisco, CA 94102",
    Address
)
print(f"Street: {address.street}")
print(f"City: {address.city}")
print(f"Postal Code: {address.postalcode}")

import marvin

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

Map with complex output types
descriptions = [
    "Software engineer, 5 years experience, New York",
    "Data scientist, fresh graduate, San Francisco",
    "Product manager, 3 years, 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:

    """Create a brief summary of the given text in 2-3 sentences."""

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

Run async
async def main():
    texts = [
        "Great product!",
        "Disappointing service",
        "It's okay",
        "Amazing! Will buy again",
        "Did not meet expectations"
    ]

    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: Define schemas and classifications
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: Create AI functions
@marvin.fn
def generatesummary(feedback: str) -> str:
    """Create a one-sentence summary of the customer feedback."""

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

    """Suggest an action to take based on customer feedback.
    Consider the category and sentiment to provide appropriate suggestions."""

@marvin.fn
def calculatepriority(feedback: str, sentiment: str) -> int:
    """Calculate a priority score of 1-10 based on feedback and sentiment.
    10 = very urgent, 1 = not urgent.
    Consider: issue severity, churn potential, business impact."""

Step 3: Main pipeline
def processfeedback(text: str) -> ProcessedFeedback:

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

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

    # Generate summary and 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]

    # Aggregate analysis
    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: Run the pipeline
feedbacks = [
    "The app keeps crashing during checkout. Happened 3 times this week!",
    "The new recommendation engine feature is fantastic, really helpful!",
    "Subscription price went up but no new features. Might switch to competitor X.",
    "Slow delivery, 5 days and still not arrived. Customer service unresponsive.",
    "The new UI redesign is cleaner and easier to use. Good job!",
    "It's average, nothing special compared to other apps.",
]

Process all feedback
analysis = analyzefeedbackbatch(feedbacks)


Display results
print(f"\n{'='60}")
print(f"CUSTOMER FEEDBACK ANALYSIS REPORT")
print(f"{'='*60}")
print(f"Total Feedback: {analysis['totalfeedbacks']}")
print(f"\nSentiment Distribution:")
for sent, count in analysis['sentimentdistribution'].items():

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

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

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

    print(f"  Category: {item.category} | Sentiment: {item.sentiment}")
    print(f"  Action: {item.suggestedaction}")

Tips and 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

Conclusion

Marvin AI is the perfect toolkit for developers who want to integrate AI into their applications without the complexity of large frameworks. With its "AI as a function" approach, Marvin makes AI engineering feel as natural as writing regular Python code.

The main advantage of Marvin over other frameworks is its simplicity. You do not need to understand concepts like chains, graphs, or agents. Just define what you want through type hints and docstrings, and Marvin handles the rest. This makes Marvin an ideal choice for rapid prototyping, microservices, and integrating AI into existing applications.

Start with simple features like classify or fn, then expand to more complex pipelines as needed. Happy building!