The cheapest way to classify1,000,000 posts

import asyncio
from typesafe_sdk import AsyncTypeSafeClient, Choice, Noul

RUBRIC = {
    "product": "A launch, a feature, a release, a changelog",
    "opinion": "A take, an argument, a prediction",
    "personal": "Life, travel, family, health",
    "promo": "Asks the reader to buy, click, sign up or subscribe",
}

async def classify(client, post):
    response = await client.system_one(
        state={"post": post["text"], "platform": "X"},
        model="jev-1.13.0",          # pin it. jev-latest moves under you.
        questions={
            "topic": Choice(instructions="What is this post mainly about", criteria=RUBRIC),
            "is_spam": Noul(instructions="This post is spam or an unsolicited advertisement"),
            "is_rage_bait": Noul(
                instructions="This post is written to provoke an angry reply rather than to inform",
            ),
        },
    )
    topic = response.answers["topic"]
    return {
        "id": post["id"],
        "topic": topic.choice,
        "topic_confidence": topic.confidence,
        "spam": response.answers["is_spam"].noul,        # a float, not a bool
        "rage_bait": response.answers["is_rage_bait"].noul,
        "input_tokens": response.usage.input_tokens,     # the only billed number
    }

async def run(posts, concurrency=20):
    sem = asyncio.Semaphore(concurrency)
    async with AsyncTypeSafeClient() as client:
        async def guarded(post):
            async with sem:
                return await classify(client, post)
        return await asyncio.gather(*(guarded(p) for p in posts))

from typesafe_sdk import Choice, Noul

# Batching posts into one request trades accuracy risk for request headroom.
# It does NOT save tokens: each post still needs its own question with its own
# instructions, so the input total is roughly unchanged. What drops is the
# request count, which is the ceiling that actually binds.

def batched_questions(posts):
    questions = {}
    state = {}
    for i, post in enumerate(posts):
        state[f"post_{i}"] = post["text"]
        questions[f"topic_{i}"] = Choice(
            instructions=f"What is post_{i} mainly about",
            criteria=RUBRIC,
        )
        questions[f"spam_{i}"] = Noul(
            instructions=f"post_{i} is spam or an unsolicited advertisement",
        )
    return state, questions

# Budget check before you send it:
#   state + longest question must stay under 32k tokens
#   the whole request must stay under 64k tokens
#
# Accuracy check you must actually run:
#   score 500 posts batched and unbatched, compare label by label.
#   TypeSafe's own docs warn that irrelevant context degrades answers,
#   and in a batch every question sees every post.

curl -X POST https://opentweet.io/api/v1/posts \
  -H "Authorization: Bearer ot_your_key" \
  -H "Content-Type: application/json" \
  -d '{
    "text": "We classified a million posts this week. 3.1% were rage bait.",
    "publish_now": true,
    "platforms": ["x", "bluesky", "linkedin"]
  }'