MAKER Workflow

95% per-step accuracy over 100 steps:
  0.95^100 = 0.6% overall success ❌

99.9% per-step accuracy (with MAKER) over 100 steps:
  0.999^100 = 90% overall success ✅

For million-step tasks:
  Even 99% per-step fails
  MAKER enables 99.99%+ per-step reliability

import asyncio
from fast_agent import FastAgent

fast = FastAgent("MAKER Example")

# Define a classifier using a cheap model (Haiku)
@fast.agent(
    name="classifier",
    model="claude-3-haiku-20240307",
    instruction="""You are a customer support intent classifier.
Classify the customer message into exactly one of: COMPLAINT, QUESTION, REQUEST, FEEDBACK.
Respond with ONLY the single word classification, nothing else.

Examples:
- "This product is broken!" → COMPLAINT
- "How do I reset my password?" → QUESTION
- "Please cancel my subscription" → REQUEST
- "Just wanted to say I love the new feature" → FEEDBACK""",
)

# Wrap with MAKER for reliable, consistent classification
@fast.maker(
    name="reliable_classifier",
    worker="classifier",
    k=3,  # Require 3-vote margin for consensus
    max_samples=10,  # Max attempts before falling back to plurality
    match_strategy="normalized",  # Ignore case/whitespace differences
    red_flag_max_length=20,  # Discard verbose responses (should be one word)
)
async def main():
    async with fast.run() as agent:
        # Classify ambiguous customer messages
        result = await agent.reliable_classifier.send("I've been waiting for 3 days now.")
        print(result)

if __name__ == "__main__":
    asyncio.run(main())

Sample 1: "COMPLAINT"     Votes: {COMPLAINT: 1}
Sample 2: "COMPLAINT"     Votes: {COMPLAINT: 2}
Sample 3: "QUESTION"      Votes: {COMPLAINT: 2, QUESTION: 1}
Sample 4: "COMPLAINT"     Votes: {COMPLAINT: 3, QUESTION: 1}
Sample 5: "COMPLAINT"     Votes: {COMPLAINT: 4, QUESTION: 1}
                           Leader margin: 4 - 1 = 3 ✅
                           Winner: "COMPLAINT" (converged)

match_strategy="exact"  # "Hello" ≠ "hello"

match_strategy="normalized"  # "Hello World" = "hello  world" = "HELLO WORLD"

match_strategy="structured"  # {"a": 1, "b": 2} = {"b": 2, "a": 1}

def custom_normalizer(response: str) -> str:
    # Extract only digits
    return "".join(c for c in response if c.isdigit())

@fast.maker(
    name="number_extractor",
    worker="extractor",
    k=3,
    match_fn=custom_normalizer,  # Overrides match_strategy
)

@fast.maker(
    name="concise_classifier",
    worker="classifier",
    k=3,
    red_flag_max_length=20,  # Expect one-word answers
)

def validate_classification(response: str) -> bool:
    valid_classes = {"COMPLAINT", "QUESTION", "REQUEST", "FEEDBACK"}
    return response.strip().upper() in valid_classes

@fast.maker(
    name="validated_classifier",
    worker="classifier",
    k=3,
    red_flag_validator=validate_classification,
)

result = await agent.reliable_classifier.send("Message to classify")

# Access detailed voting statistics
stats = agent.reliable_classifier.last_result
print(f"Winner: {stats.winner}")
print(f"Votes: {stats.votes}")  # e.g., {'COMPLAINT': 4, 'QUESTION': 1}
print(f"Total samples: {stats.total_samples}")
print(f"Discarded samples: {stats.discarded_samples}")
print(f"Winning margin: {stats.margin}")
print(f"Converged: {stats.converged}")  # True if k-margin achieved

@fast.agent(
    "email_validator",
    model="haiku",
    instruction="""Validate if the input is a properly formatted email address.
Respond with only: VALID or INVALID""",
)
@fast.maker(
    name="reliable_email_validator",
    worker="email_validator",
    k=5,  # Very high reliability for data validation
    max_samples=15,
    match_strategy="normalized",
    red_flag_max_length=10,
)

async def validate_million_emails(emails: list[str]) -> list[bool]:
    results = []
    async with fast.run() as agent:
        for email in emails:
            result = await agent.reliable_email_validator.send(email)
            results.append(result == "VALID")
    return results

@fast.agent(
    "syntax_checker",
    model="gpt-3.5-turbo",  # Cheap model
    instruction="""Check if the Python code has syntax errors.
Respond with only: VALID or ERROR""",
)
@fast.maker(
    name="reliable_syntax_checker",
    worker="syntax_checker",
    k=3,
    max_samples=12,
    match_strategy="normalized",
)

async def check_codebase(files: list[str]) -> dict[str, bool]:
    results = {}
    async with fast.run() as agent:
        for file_path in files:
            with open(file_path) as f:
                code = f.read()
            result = await agent.reliable_syntax_checker.send(
                f"Check syntax: {code}"
            )
            results[file_path] = result == "VALID"
    return results

@fast.agent(
    "data_extractor",
    model="haiku",
    instruction="""Extract name, email, and phone from the text.
Respond with JSON: {"name": "...", "email": "...", "phone": "..."}""",
)
@fast.maker(
    name="reliable_extractor",
    worker="data_extractor",
    k=4,
    max_samples=20,
    match_strategy="structured",  # Compare parsed JSON
    red_flag_max_length=200,
)

async def extract_contact_info(documents: list[str]) -> list[dict]:
    results = []
    async with fast.run() as agent:
        for doc in documents:
            result = await agent.reliable_extractor.send(doc)
            results.append(json.loads(result))
    return results

Typical Convergence (k=3, 95% per-step accuracy):
- Average samples: 5-7
- Convergence rate: 90%+
- Effective accuracy: 99.9%+

Cost Example:
- Worker: Haiku @ $0.25/MTok
- Average 6 samples per task
- 1000 tasks = 6000 calls
- Still cheaper than 1000 calls to expensive model

import logging
logging.basicConfig(level=logging.DEBUG)

# You'll see:
# DEBUG: Sample 1: 1 votes for this response
# DEBUG: Sample 2: 2 votes for this response
# DEBUG: Sample 3: 1 votes for alternative response
# DEBUG: Sample 4: 3 votes for this response
# DEBUG: MAKER converged: 3 votes, margin 2, 4 samples