直播聊天过滤：辱骂、剧透、垃圾消息一次拦下

💡 一句话总结：用 Jev 建立 AI 直播弹幕过滤流水线 —— 一条弹幕一次请求三道题、8 线程并发处理、纯代码规则决定观众看到什么。

你在看直播，弹幕刷得比你能读的还快。官方 Twitch chat filter cookbook 的做法：每条消息到达时单独发给 Jev 一次请求，带上直播间标题和游戏名，问三道题——这是哪类消息、是否辱骂、是否剧透；再用几条朴素规则把答案变成观众看到的：提问高亮，辱骂、指点、广告隐藏，剧透打码。

关键词列表能抓住垃圾链接，但「go left, there's a site of grace behind the tree」里没有一个坏词，剧情反转也没有——这正是 AI 直播弹幕过滤和黑名单的分水岭：判断在语义里，不在字面。官方用 29 条艾尔登法环直播间弹幕实测：隐藏 7 条、打码 2 条、高亮 6 条提问，请求中位耗时约 0.1 秒。本课约 15 分钟带你整套跑通，而且不设 API Key 也能复现官方全部数字——第 1 步会给你官方缓存文件。

学完你能做什么

• 搭一条 AI 直播弹幕过滤流水线：每条弹幕一次请求三道题，ThreadPoolExecutor 8 线程并发跑完整场聊天
• 读懂三道题的分工：category Choice 定默认处置，toxic / spoiler 两道 Noul 负责一票否决和打码
• 用 decide() 把答案变成动作：改规则只改代码，29 份答案原样复用，0 次新请求
• 不花一分钱复现官方数字：下载官方 json_cache.json 缓存重放；换自己的弹幕前先校准阈值

为什么用 Jev，而不是关键词黑名单？

黑名单是现状，普通 LLM 是另一个极端。官方 cookbook 开篇就点出黑名单的局限：关键词能抓垃圾链接，但抓不住没有坏词的剧透和指点。把三种方案摆在一起看：

能力	关键词黑名单	普通 LLM	Jev
没有脏词的剧透/指点	❌ 「树后面有个赐福」不含任何违禁词，无感	✅ 读得懂，但答案是一段要再解析的文本	✅ 官方实测：指点 backseat 1.00、剧透 spoiler 0.95，概率直接进代码
玩笑和辱骂的分寸	❌ 「you're so bad at this boss」含 bad，朋友间玩笑也挨刀	⚠️ 说得出区别，「有多确定」要追问	✅ 玩笑式嘲讽 toxic 0.28、真辱骂 0.97，一刀两断
高峰期吞吐	✅ 匹配快	❌ 逐条生成回复，弹幕不等它	✅ 官方实测请求中位约 0.1 秒，8 线程并发
改规则	✅ 改表即生效	❌ 规则写在 prompt 里，一改就要全量重跑	✅ 规则在 decide() 里，29 份答案复用，0 新请求
答案形态	✅ 布尔	❌ 自由文本或 JSON，解析会失败	✅ 类型化答案 + 概率分布，无解析

说句公道话：黑名单该留着——抓垃圾链接它又快又便宜，Jev 补的是它看不见的那一半。两者不冲突。

你现在的困境

• 「开播三小时，骂人的、剧透的、卖粉的混在弹幕里刷，人工盯屏盯到眼花，聊天审核永远慢半拍」
• 「关键词黑名单只会匹配字面：『往左走，树后面有个赐福』没有一个违禁词，照样把首播体验剧透光」
• 「想上 LLM 做 AI 评论过滤，一条弹幕等好几秒才有回音，高峰期直接堆死；让它只回标签，同一句话两次答案还不一样」

什么时候用这一招

当你需要：

• 弹幕到达后立刻出判定，高峰期也不断流
• 拦的不只是脏词：剧透、未经请求的游戏指点、广告，都按语义判断
• 规则随时改——今天藏广告，比赛日连欢呼也藏，而且改规则不重新花钱

而且不想：

• 维护一张永远追不上变体的关键词表
• 每条弹幕解析一遍 LLM 吐出来的 JSON
• 剧透判断脱离游戏上下文——同一句话在不同游戏里含义完全不同

🎒 开始前的准备

• 完成了 Choice 选择题：会写 criteria 选项 map，读得懂 choice 和 probabilities
• 完成了 Noul 是非题：知道 noul 是 0–1 概率、起点阈值 yes ≥ 0.7 / no ≤ 0.3
• 跑通过安装与第一个请求：.venv 可用（本页支持缓存重放，暂时没有 Key 也行，见第 1 步）
• 一杯水 ☕——本课是一个完整小项目：29 条弹幕 + 三道题 + 一套能跑的过滤代码

跟我做

先说好怎么跑：新建 twitch_chat_filter.py，把下面每一步的代码块按顺序贴进同一个文件；每完成一步就运行一次 python twitch_chat_filter.py。答案会被缓存，重复运行不重复花钱。

第 1 步：装环境、下官方缓存，备好 29 条弹幕样本

为什么：官方为这个案例写了一整套真实感十足的弹幕——艾尔登法环直播间，29 条，每类消息至少出现两次（含两条纯表情）。我们直接用它验证整套过滤逻辑，不用等真实直播。官方还发布了这 29 条的答案缓存 json_cache.json，放对位置就能不调 API 复现官方全部数字。

安装依赖。cooksafe 是官方 cookbook 的配套小工具库（PyPI 自述「Shared helpers for the TypeSafe cookbooks」），本页用到它的 JsonCache 缓存和 make_playground_link 分享链接：

pip install 'typesafe-sdk>=0.7.1,<0.8.0' ipython 'cooksafe>=0.2.0,<0.3.0'

下载官方缓存（raw 直链，实测可用。不下也行——设了 TYPESAFE_API_KEY 就是真请求，花费见模型与价格）：

curl -sL -o json_cache.json https://raw.githubusercontent.com/typesafe-ai/typesafe-public-examples/main/cookbooks/twitch_chat_filter/json_cache.json

新建 twitch_chat_filter.py，贴入第一块代码——客户端、缓存和样本数据。注意 api_key 的默认值 "cache-only"：没有 Key 时，缓存重放照常工作：

from concurrent.futures import ThreadPoolExecutor
import os
from pathlib import Path
from time import perf_counter

from cooksafe import JsonCache, make_playground_link
from IPython.display import Markdown, display
from typesafe_sdk import Choice, Noul, NoulCriteria, TypeSafeClient

TYPESAFE_MODEL = "jev-1.14.0"  # 官方数字的采样版本（2026-10-29）

client = TypeSafeClient(
    api_key=os.environ.get("TYPESAFE_API_KEY", "cache-only"),  # cached re-renders need no key
    base_url=os.environ.get("TYPESAFE_ENDPOINT"),
    timeout=60.0,
)
CACHE_DIR = Path(__file__).resolve().parent if "__file__" in globals() else Path.cwd()
json_cache = JsonCache(CACHE_DIR / "json_cache.json")

STREAM = {"title": "Elden Ring first playthrough! No spoilers please", "game": "Elden Ring"}

CHAT = [
    ("night-owl-22", "hi chat, first time catching a stream live"),
    ("ggwp-andy", "PogChamp PogChamp PogChamp"),
    ("mira-plays", "how many hours have you played so far?"),
    ("tarnished-tom", "this area looks so good at night"),
    ("backseat-bert", "go left, there's a site of grace behind the tree"),
    ("follow4follow-bot", "Want more viewers? Buy 1000 followers for $5 at cheapviewz dot example"),
    ("lore-hound", "wait till you find out Ranni planned the Night of Black Knives"),
    ("kappa-kev", "LUL LUL LUL"),
    ("sunny-skies", "are you going to do a no-hit run after this?"),
    ("rage-ryan", "you're genuinely the worst player i've ever seen, uninstall trash"),
    ("cozy-cat", "the music in this area is so good"),
    ("backseat-bert", "use your flask NOW, why are you not healing"),
    ("mira-plays", "W streamer"),
    ("ggwp-andy", "GG that was insane"),
    ("quiet-quinn", "anyone else watching from Brazil?"),
    ("tarnished-tom", "what build are you going for?"),
    ("lore-hound", "the ending where you become Elden Lord with Ranni is the best one"),
    ("kappa-kev", "chat is wild today lol"),
    ("backseat-bert", "the boss is weak to fire, equip the torch before you go in"),
    ("rage-ryan", "nobody cares about your opinion, go back to your basement"),
    ("cozy-cat", "you're so bad at this boss lmao, love you tho"),
    ("sunny-skies", "who is the final boss?"),
    ("ggwp-andy", "LETS GOOOO"),
    ("night-owl-22", "i just got this game too, it's so hard"),
    ("promo-pete", "check out my channel, I stream every day, follow me!"),
    ("quiet-quinn", "dodge rolling through that was clean"),
    ("mira-plays", "is the DLC worth it?"),
    ("cozy-cat", "your character looks so cool"),
    ("night-owl-22", "this boss took me 40 tries lol"),
]

print(f"{len(CHAT)} messages. What one request sends as its state:")
print({"stream": STREAM, "message": {"user": CHAT[4][0], "text": CHAT[4][1]}})

运行 python twitch_chat_filter.py。

你应该看到：

29 messages. What one request sends as its state:
{'stream': {'title': 'Elden Ring first playthrough! No spoilers please', 'game': 'Elden Ring'}, 'message': {'user': 'backseat-bert', 'text': "go left, there's a site of grace behind the tree"}}

两个设计点先记下。直播标题和游戏名进每一次请求：一句话是不是 backseat 指点、是不是剧透，取决于在玩哪款游戏，所以它们是 state 的一部分。每条消息一个请求：消息之间互不依赖，这为第 5 步的并发铺路。

• 我跑出了 29 messages 这两行，json_cache.json 就在脚本旁边

第 2 步：写出三道问题：一道 Choice 分类加两道 Noul

为什么：一条弹幕要判断三件事——它是什么（决定默认处置）、是否辱骂（一票否决）、是否剧透（打码）。三道题对同一份 state 一次请求发出，Jev 并行、独立评估，加题几乎不增加响应时间。

接着贴入问题定义：

QUESTIONS = {
    "category": Choice(
        instructions="Which category best describes the main purpose of `message.text`?",
        criteria={
            "question": "A genuine request for information from the streamer or chat, "
            "not gameplay instructions phrased as a question or a promotional pitch",
            "hype": "Cheering, reactions, or repeated emotes (PogChamp, LUL, GG, W, "
            "lets gooo); repetition alone does not make cheering spam",
            "conversation": "Ordinary chatter, opinions, replies, useful shared links, "
            "or any message whose main purpose does not fit the other categories",
            "backseat": "Unrequested gameplay advice telling the streamer what to do, "
            "including instructions phrased as a question; not a request for information",
            "spam": "Unsolicited advertising, self-promotion, follower selling, or "
            "promotional links; not ordinary cheering, repeated emotes, or a useful link",
        },
    ),
    "toxic": Noul(
        instructions="Is this message insulting, hateful or harassing towards someone?",
        criteria=NoulCriteria(
            true="Insults, slurs, threats or personal attacks on the streamer or another viewer",
            false="Friendly, neutral, or light banter with no target",
        ),
    ),
    "spoiler": Noul(
        instructions="Does this message reveal a plot point or ending of the game being played?"
    ),
}

五个类目的边界是这套 rubric 的灵魂，官方原文划了三条最容易踩混的线：

• question ≠ 指令：真心求信息才算；伪装成提问的游戏指点不算
• hype ≠ spam：刷表情、欢呼是 hype——官方原话「repetition alone does not make cheering spam」，重复本身不等于垃圾
• spam ≠ 有用链接：未经请求的广告、卖粉、推广链接才是 spam——官方原话「A link alone is not spam」
• conversation 是兜底：其他类目都套不上时落这里

toxic 的 NoulCriteria 把玩笑划了出去：true 是侮辱、歧视、威胁、人身攻击，false 是「友好、中性、无对象的轻玩笑」——第 3 步你会看到这条边界救下一条看起来像辱骂的消息。spoiler 只写 instructions，官方没给它定义 criteria。

你应该看到：运行还是第 1 步那两行——问题定义只是数据，还没发请求。确认你文件里的 QUESTIONS 和上面一字不差：缓存键由 state 和问题共同决定（第 3 步细说），抄错一个字就会 miss 缓存、变成真请求。

第 3 步：单条弹幕判定：一次请求拿到三个答案

为什么：先用三条最典型的弹幕验证问题写得好不好——纯指点、剧情剧透、玩笑式嘲讽各一条。classify() 组装 state 并发出请求；_ask() 被 @json_cache 装饰：答案按「state + 问题写成 JSON」的键存进 json_cache.json，重跑直接复用。

接着贴入：

def classify(user: str, text: str, stream: dict | None = None) -> dict:
    state = {
        "stream": STREAM if stream is None else stream,
        "message": {"user": user, "text": text},
    }
    questions = {key: q.model_dump(mode="json", exclude_none=True) for key, q in QUESTIONS.items()}
    return _ask(state, questions)


@json_cache
def _ask(state: dict, questions: dict) -> dict:
    if not os.environ.get("TYPESAFE_API_KEY"):
        raise RuntimeError(
            "Set TYPESAFE_API_KEY for a live request, or put the cookbook's "
            "json_cache.json beside your script (in the working directory for a notebook)."
        )
    started = perf_counter()
    response = client.system_one(state=state, questions=questions, model=TYPESAFE_MODEL)
    category = response.answers["category"]
    return {
        "category": category.choice,
        "probabilities": dict(category.probabilities),
        "toxic": response.answers["toxic"].noul,
        "spoiler": response.answers["spoiler"].noul,
        "model": response.model,
        "seconds": round(perf_counter() - started, 2),
    }


def show(user: str, text: str) -> None:
    result = classify(user, text)
    print(f"{user}: {text}")
    for label, p in sorted(result["probabilities"].items(), key=lambda kv: -kv[1]):
        print(f"  {label:<13}{p:5.2f}  {'#' * round(p * 30)}".rstrip())
    print(f"  {'toxic':<13}{result['toxic']:5.2f}")
    print(f"  {'spoiler':<13}{result['spoiler']:5.2f}\n")


show("backseat-bert", "go left, there's a site of grace behind the tree")
show("lore-hound", "wait till you find out Ranni planned the Night of Black Knives")
show("cozy-cat", "you're so bad at this boss lmao, love you tho")

运行。

你应该看到（官方缓存重放的数字；换成真请求，数值会有小差异）：

backseat-bert: go left, there's a site of grace behind the tree
  backseat      1.00  ##############################
  question      0.00
  hype          0.00
  conversation  0.00
  spam          0.00
  toxic         0.03
  spoiler       0.07

lore-hound: wait till you find out Ranni planned the Night of Black Knives
  conversation  0.67  ####################
  backseat      0.32  ##########
  hype          0.01
  question      0.00
  spam          0.00
  toxic         0.04
  spoiler       0.95

cozy-cat: you're so bad at this boss lmao, love you tho
  conversation  0.79  ########################
  hype          0.21  ######
  question      0.00
  backseat      0.00
  spam          0.00
  toxic         0.28
  spoiler       0.02

三条读法：

• 指点被钉死：backseat-bert 拿到 backseat 1.00，同时 toxic 只有 0.03——它烦人但不辱骂。两个维度分开判，这就是拆题的价值
• 剧透靠 spoiler 抓住：lore-hound 这句在 conversation 0.67 和 backseat 0.32 之间摇摆，类目怎么争都没用——spoiler 0.95 直接触发打码
• 玩笑被边界救下：cozy-cat 这句读着像辱骂，但结尾的「love you tho」落进了 false criteria 的「轻玩笑」——toxic 0.28，远低于阈值，照常显示

💡 缓存的键：改题必失效

缓存键是 state 和问题写成的 JSON——编辑任何一道题都会换键，下次运行就是真请求而不是重放旧答案。键里不含模型：想换 TYPESAFE_MODEL 跑新的，把缓存文件挪开再跑。没有 Key 又 miss 了缓存？_ask 会抛出 RuntimeError 提示你：设 Key，或把官方 json_cache.json 放回脚本旁边。

• 三条示例的输出和上面一致（用官方缓存时）；我能指出哪几行是类目概率、哪两行是 Noul

第 4 步：写 decide()：三条检查按顺序把答案变成动作

为什么：判定逻辑是纯代码，不发任何请求。官方的顺序有讲究：先辱骂（一票否决，无论类目），再类目规则，最后剧透（打码，已被隐藏的不重复处理）。

接着贴入：

RULES = {
    "question": "highlight",
    "hype": "show",
    "conversation": "show",
    "backseat": "hide",
    "spam": "hide",
}
TOXIC_AT = 0.7  # hide at or above this chance of toxic
SPOILER_AT = 0.7  # blur at or above this chance of a spoiler


def decide(result: dict, rules: dict = RULES) -> str:
    if result["toxic"] >= TOXIC_AT:
        return "hide (toxic)"
    action = rules[result["category"]]
    if action == "hide":
        return f"hide ({result['category']})"
    if result["spoiler"] >= SPOILER_AT:
        return "blur (spoiler)"
    return action

关于 0.7 这两个阈值，官方有个重要观察：这 29 条样本里，toxic 概率要么 ≥ 0.97 要么 ≤ 0.38，spoiler 要么 ≥ 0.95 要么 ≤ 0.38——中间是空的。所以在这套样本上，0.38 到 0.95 之间取任何阈值结果都一样。但你自己的频道大概率会有落在中间的值，官方原话：上线前拿自己的弹幕把两个阈值都校准一遍。

你应该看到：还没有新输出——decide() 要等下一批结果才有东西可判。先口头过一遍三条检查：toxic ≥ 0.7 → hide；类目规则是 hide → hide；spoiler ≥ 0.7 → blur；剩下的按类目规则走（show / highlight / dim）。

• 我能说出三条检查的顺序，以及为什么 toxic 排第一

第 5 步：并发跑完 AI 直播弹幕过滤，输出观众视角

为什么：弹幕不等人，29 条消息互相独立——线程池 8 个并发一次处理，pool.map 按聊天顺序返回答案。第一次真跑会发 26 个新请求（3 条已被第 3 步缓存）；用官方缓存则一个请求都不发。

接着贴入：

with ThreadPoolExecutor(max_workers=8) as pool:  # 29 requests, one per message, 8 at once
    results = list(pool.map(lambda message: classify(*message), CHAT))

counts: dict[str, int] = {}
for (user, text), result in zip(CHAT, results):
    action = decide(result)
    counts[action] = counts.get(action, 0) + 1
    if action.startswith("hide"):
        line = f"   · hidden, {action[6:-1]}"
    elif action == "blur (spoiler)":
        line = f"   {user}: ░░░░░░░░ spoiler"
    else:
        mark = {"highlight": ">>", "dim": "..", "show": "  "}[action]
        line = f"{mark} {user}: {text}"
    print(f"{line[:78]:<80}{result['category']}")

print()
for action, n in sorted(counts.items(), key=lambda kv: -kv[1]):
    print(f"{action:<20}{n:>3}")

运行。

你应该看到（观众视角：隐藏的只显示原因，打码的只显示作者，高亮的问题带 >>；行尾是类目答案）：

   night-owl-22: hi chat, first time catching a stream live                     conversation
   ggwp-andy: PogChamp PogChamp PogChamp                                        hype
>> mira-plays: how many hours have you played so far?                           question
   tarnished-tom: this area looks so good at night                              conversation
   · hidden, backseat                                                           backseat
   · hidden, spam                                                               spam
   lore-hound: ░░░░░░░░ spoiler                                                 conversation
   kappa-kev: LUL LUL LUL                                                       hype
>> sunny-skies: are you going to do a no-hit run after this?                    question
   · hidden, toxic                                                              conversation
   cozy-cat: the music in this area is so good                                  conversation
   · hidden, backseat                                                           backseat
   mira-plays: W streamer                                                       hype
   ggwp-andy: GG that was insane                                                hype
>> quiet-quinn: anyone else watching from Brazil?                               question
>> tarnished-tom: what build are you going for?                                 question
   lore-hound: ░░░░░░░░ spoiler                                                 conversation
   kappa-kev: chat is wild today lol                                            conversation
   · hidden, backseat                                                           backseat
   · hidden, toxic                                                              conversation
   cozy-cat: you're so bad at this boss lmao, love you tho                      conversation
>> sunny-skies: who is the final boss?                                          question
   ggwp-andy: LETS GOOOO                                                        hype
   night-owl-22: i just got this game too, it's so hard                         conversation
   · hidden, spam                                                               spam
   quiet-quinn: dodge rolling through that was clean                            hype
>> mira-plays: is the DLC worth it?                                             question
   cozy-cat: your character looks so cool                                       conversation
   night-owl-22: this boss took me 40 tries lol                                 conversation

show                 14
highlight             6
hide (backseat)       3
hide (spam)           2
blur (spoiler)        2
hide (toxic)          2

对账：29 条 = 显示 14 + 高亮 6 + 隐藏 7（指点 3、广告 2、辱骂 2）+ 打码 2，和官方开头说的「hides seven, blurs two, highlights six」严丝合缝。两个细节：rage-ryan 的辱骂类目其实是 conversation，但 toxic 检查排在类目规则前面，照样拦下；lore-hound 的两条剧透类目也是 conversation，靠第三条 spoiler 检查打码——三条 if 的顺序，就是处置的优先级。

⚠️ 中文弹幕先测再上线

官方说明：这个 cookbook 发布时，Jev 主要用英语训练——其他语言能用但准确度较低。B 站、视频号这类中文弹幕场景，上线前先拿自己频道的真实弹幕跑一批，重点看概率分布，再定阈值。语言支持的完整说明见模型与价格。

• 29 行观众视角 + 7 行计数和上面一致；隐藏合计 7 条

第 6 步：改规则零新请求：把欢呼也藏掉

为什么：直播的风格会变——比赛日要安静，互动日要热闹。规则是数据不是 prompt：答案只描述消息本身，换一套 RULES 就能复用第 5 步的 29 份答案，一次新请求都不发。

接着贴入：

QUIET = RULES | {"hype": "hide", "conversation": "dim"}

changed = 0
for (user, text), result in zip(CHAT, results):
    before, after = decide(result), decide(result, QUIET)
    if before != after:
        changed += 1
        print(f"{user + ': ' + text:<62.62}  {before:<10} -> {after}")
print(f"\n{changed} messages changed, 0 new requests")

运行。

你应该看到：

night-owl-22: hi chat, first time catching a stream live        show       -> dim
ggwp-andy: PogChamp PogChamp PogChamp                           show       -> hide (hype)
tarnished-tom: this area looks so good at night                 show       -> dim
kappa-kev: LUL LUL LUL                                          show       -> hide (hype)
cozy-cat: the music in this area is so good                     show       -> dim
mira-plays: W streamer                                          show       -> hide (hype)
ggwp-andy: GG that was insane                                   show       -> hide (hype)
kappa-kev: chat is wild today lol                               show       -> dim
cozy-cat: you're so bad at this boss lmao, love you tho         show       -> dim
ggwp-andy: LETS GOOOO                                           show       -> hide (hype)
night-owl-22: i just got this game too, it's so hard            show       -> dim
quiet-quinn: dodge rolling through that was clean               show       -> hide (hype)
cozy-cat: your character looks so cool                          show       -> dim
night-owl-22: this boss took me 40 tries lol                    show       -> dim

14 messages changed, 0 new requests

14 条消息改判——6 条欢呼转隐藏、8 条闲聊转调暗——0 次新请求。RULES | {...} 是字典合并（Python 3.9+；SDK 本来就要求 3.10+）。想换回默认规则，decide(result) 不传第二参就行。

💡 这就是 Jev 的独有价值

三道题的答案只描述消息（「这是指点」「辱骂概率 0.97」），不描述动作——动作全在你的 decide() 里。所以改规则零请求、A/B 两套规则零成本、概率原生不需要解析 JSON。普通 LLM 把规则写进 prompt，改一个字就要对每条消息重新推理一遍。

• 我跑出 14 messages changed, 0 new requests，且没有产生新的 API 花费

📋 魔法 Prompt

本页的「Prompt」是给 API 的问题模板——抄走、替换【】占位符（带 ? 的是选填）就能用到你自己的直播间。三件套：分类题、双 Noul、规则表。

📝 弹幕分类题（Choice 版）

预期效果：把一个直播间的聊天分成五个互斥类目，一次请求拿到完整概率分布，直接接你的规则表。

from typesafe_sdk import Choice

# 必填项：【】里换成你直播间的真实边界
CHAT_CATEGORY = Choice(
    instructions="【一句话问题：这条弹幕的主要目的是什么？引用 `message.text`】",
    criteria={
        "question": "【真心求信息的情形；写明「伪装成提问的指令/推广」不算】",
        "hype": "【欢呼、反应、刷表情的情形；写明「重复本身不等于垃圾」】",
        "conversation": "【日常闲聊、观点、有用链接；其他类目都套不上时落这里】",
        "backseat": "【未经请求的游戏/操作指点，含伪装成提问的指令】",
        "spam": "【未经请求的广告、卖粉、推广链接；写明「有用链接不算」】",
    },
)

# 输出规范：answer.choice 是最高概率类目；answer.probabilities 是五个类目的完整分布
# 约束条件：
# ✅ 一个类目一个维度，互斥且合起来覆盖常见弹幕
# ✅ 边界写进选项描述：hype 和 spam、question 和 backseat 最容易混
# ❌ 不要把「隐藏」「打码」这类动作写进类目——动作属于规则表

✅ 辱骂与剧透双 Noul（Noul 版）

预期效果：两道一票否决题各拿到 0–1 概率：辱骂达标直接隐藏，剧透达标打码；边界模糊的加 NoulCriteria 钉死。

from typesafe_sdk import Noul, NoulCriteria

# 必填项：问句方向要正，yes = 命中你要检查的事
TOXIC_QUESTION = Noul(
    instructions="【这条弹幕是否辱骂、仇恨或骚扰某人？】",
    criteria=NoulCriteria(
        true="【侮辱、歧视、威胁、人身攻击的情形】",
        false="【友好、中性、无对象的轻玩笑——把要放过的玩笑写清楚】",
    ),
)
SPOILER_QUESTION = Noul(
    instructions="【这条弹幕是否剧透正在玩的游戏的剧情或结局？】",
    # 选填项：剧透边界模糊时补 criteria?
    # criteria=NoulCriteria(true="【…】", false="【…】"),
)

# 输出规范：answer.noul 是 0–1 概率，没有 confidence；接近 0.5 = 不确定
# 约束条件：
# ✅ 一道 Noul 只问一个是非；「多严重」交给 Score 题
# ✅ 玩笑和辱骂的分界写进 false，不然熟人间互怼会被误杀
# ❌ 不要拿 0.5 当阈值——0.49 和 0.51 会触发相反动作

🔧 处置规则与阈值（decide 版）

预期效果：把类目映射成观众看到的动作，两道 Noul 各配一个阈值——改这张表就是改产品行为，零新请求。

# 必填项：【】里换成你的处置策略
RULES = {
    "question": "【highlight / show】",
    "hype": "【show / hide】",
    "conversation": "【show / dim】",
    "backseat": "【hide】",
    "spam": "【hide】",
}
TOXIC_AT = 【0.7 起步，官方示例值；用你的弹幕校准】
SPOILER_AT = 【0.7 起步，官方示例值；用你的弹幕校准】

# 输出规范：decide(result) -> "show" / "highlight" / "dim" / "hide (…)" / "blur (spoiler)"
# 约束条件：
# ✅ 顺序固定：先 toxic 一票否决，再类目规则，最后 spoiler 打码
# ✅ 阈值上线前用自己的弹幕校准——官方样本里 0.38–0.95 之间是空的，你的不会
# ❌ 不要在规则表里处理「不确定」——那是概率出口的活，见自一致性课

检查点 ✅

• twitch_chat_filter.py 跑出 29 行观众视角 + 计数：show 14 / highlight 6 / 隐藏 7 / blur 2
• 不设 TYPESAFE_API_KEY、放好官方 json_cache.json，我复现出了官方全部数字
• 我能说出三条检查的顺序：toxic 一票否决 → 类目规则 → spoiler 打码
• QUIET 规则跑出 14 messages changed, 0 new requests
• 我知道 0.7 是示例阈值——官方样本 0.38–0.95 之间为空，我的频道要先校准再上线

踩坑提醒

❌ 串行一条条发请求，高峰期跟不上

官方明说：一次一个请求跟不上忙碌的聊天。就算中位 0.1 秒，新弹幕也会先出现在聊天里、答案才回来。官方给两个选项：新消息先隐藏，答案回来再决定显不显示；或者接受辱骂消息在屏幕上停留那零点几秒。并发照第 5 步的线程池写法。

❌ 阈值照抄 0.7 就上线

0.7 在官方样本上好用，是因为那 29 条的概率两极分化（toxic ≥ 0.97 或 ≤ 0.38，spoiler ≥ 0.95 或 ≤ 0.38），中间是空的。你自己的弹幕大概率会有 0.5 上下的值——官方原话：先把两个阈值对着自己频道的聊天校准，再上直播。阈值怎么按风险调，参考 Confidence 置信度的三档用法。

❌ 赢家标签直接执行，不看概率差距

官方提醒：类目赢家也可能是没把握的赢家——前两名概率挨得近、Noul 贴着 0.5 时，自动隐藏可能误杀。官方建议先定一个策略：这类消息转人工复核，或先保持可见，别自动隐藏；拿自己频道的弹幕测过再定。给审核系统加「不确定」出口的完整做法，见自一致性。

❌ 剧透检查不知道主播玩到哪了

spoiler 只拿游戏名和模型对这款游戏的了解判断「是不是剧情揭露」，它不知道主播和观众看没看过。想只拦「还没被剧透过的」剧情点，官方的做法：把当前进度和相关剧情信息放进 state，再把问题改成引用这些上下文。

进阶技巧（可选）

多频道一次跑。 每个频道把各自的 stream 传给 classify()——classify(user, text, stream=...) 的第三参就是干这个的；标题和游戏名不同，剧透和指点的判断就不同。

接真实聊天流。 官方建议：把平台的消息 ID（或消息页面元素的引用）跟请求一起存，答案回来时更新对应的那条消息。撞上限流就把 max_workers 调低重跑——已完成的答案都在缓存里，不会重花钱。

免代码先试。 把 state 和三道题填进 Playground（https://console.typesafe.ai/playground）就能体验：官方 cookbook 页面上有做好的分享链接，点开就是 backseat 那条消息 + 三道题。想生成自己的链接，贴入这块代码：

playground_link = make_playground_link(
    {"stream": STREAM, "message": {"user": CHAT[4][0], "text": CHAT[4][1]}},
    QUESTIONS,
)
# Print a usable URL in a script; render a clickable link in a notebook.
if __name__ == "__main__" and "__file__" in globals():
    print(f"Open in the TypeSafe playground:\n{playground_link}")
else:
    display(Markdown(f"🔗 [Open in the TypeSafe playground]({playground_link})"))

看官方实拍。 官方还有一个 Jev 驱动的 Chrome 扩展过滤真实 Twitch 直播聊天的演示视频：Twitch chat filtering Chrome extension demo。

本课小结

✅ AI 直播弹幕过滤流水线：每条弹幕一次请求三道题（category Choice + toxic / spoiler 双 Noul），8 线程并发，纯代码规则决定观众看到什么。

✅ 判定顺序：toxic ≥ 0.7 一票否决 → 类目规则（backseat / spam 隐藏、question 高亮）→ spoiler ≥ 0.7 打码。

✅ 官方实测 29 条弹幕：显示 14、高亮 6、隐藏 7（指点 3 + 广告 2 + 辱骂 2）、打码 2；请求中位约 0.1 秒。

✅ 规则是数据：QUIET 演示改判 14 条、0 次新请求；官方 json_cache.json 让你不设 Key 复现全部数字。

关键配置字段回顾：

字段	本页的值	说明
TYPESAFE_MODEL	"jev-1.14.0"	官方数字的采样版本（2026-10-29）；缓存键不含模型
TOXIC_AT / SPOILER_AT	0.7 / 0.7	示例阈值；官方样本 0.38–0.95 之间为空，自己频道先校准
RULES	question→highlight；hype/conversation→show；backseat/spam→hide	动作表；改它零新请求
max_workers	8	并发数；限流时调低，已完成的答案有缓存
每条消息问题数	3	一次请求并行评估；改任何一道题都会换缓存键

相关教程：

• LLM 护栏：输入输出全安检——同一套「概率 + 阈值」思路，这次拦的是进出 LLM 应用的消息
• 自一致性：给审核系统加「不确定」出口——前两名概率挨得近时别自动执行，给个转人工出口
• 并行提问与投机扇出——一次请求多道题的完整展开：13 题 12.2 倍便宜、10.0 倍快

下一课预告

弹幕要拦的是坏消息，分类要分的是好内容——而且类目一层比一层深。下一课：深层级分类：把内容分到类目最深处。

你会学到：

• 并行 beam search 沿 Choice 概率逐层下走，把文档分到专利、零售、生物医学、源码类目的最深处
• 类目单级装不下时怎么拆成多次请求，上一层的答案怎么决定下一层的选项
• 和本页同款思路——概率驱动代码分支，只是这次分支走的是类目树