文档目录
Twitch 聊天过滤器
在 Twitch 聊天中隐藏侮辱、垃圾信息和未经请求的游戏攻略，对剧透内容进行模糊处理，并把问题高亮显示。每条消息对应一个包含多个问题的 TypeSafe 请求，最多可并发处理八条消息，规则全部通过代码来定义。
非官方中文翻译 · 内容以 英文原文 ↗ 为准 · 同步于 2026-10-15
你正在观看一场 Twitch 直播，聊天刷得太快根本看不过来。本实战示例会对每条到达的消息进行分类：把问题高亮显示，把垃圾信息、侮辱和未经请求的游戏攻略隐藏起来，并对剧透内容进行模糊处理。一份关键词列表或许能拦下垃圾链接，但「走左边，那棵树后面有个宝箱」里没有任何违规词，剧情反转发言也一样。
把每条消息各自放进一次 Jev 请求，连同直播标题和当前游戏一起发送。我们会针对它提出三个问题：消息属于什么类型、是否带攻击性、是否会剧透游戏。少量代码就能把这些答案转化为观众看到的效果。 下面这段视频演示展示了一个由 Jev 驱动的 Chrome 扩展在实时过滤 Twitch 聊天：
本实战示例涵盖了如何用 Jev 对 29 条示例聊天消息进行分类与过滤。其中 7 条被隐藏，2 条被模糊处理，6 条被高亮为问题。
最终，你将拥有 classify()（它会对一条消息提出三个问题）以及 decide()（把答案转化为显示、高亮、变暗、隐藏或模糊处理）。当你修改某条规则（例如「如果属于 spam 则隐藏」），只有 decide() 会再次运行，不需要再向 Jev 发起请求。如果想要过滤 Twitch 或其他平台上的实时聊天，可以把这段逻辑接入自定义聊天查看器、审核工具或浏览器扩展，并按你的直播场景调整消息输入与问题。
环境准备
pip install 'typesafe-sdk>=0.7.1,<0.8.0' ipython 'cooksafe>=0.2.0,<0.3.0'新建一个名为 twitch_chat_filter.py 的文件。跟随本实战示例时，按顺序把每个 Python 代码块复制到该文件中。若想实跑，请在环境中设置 TYPESAFE_API_KEY。把所有代码块都加进去后，在该文件所在目录下运行下面这条命令：
python twitch_chat_filter.py每个 API 响应都会保存到 json_cache.json 中，与脚本放在同一目录。把这个缓存放到那里，无需再次调用 API 就能复现本文给出的数值。在 notebook 环境下，把缓存放到 notebook 的工作目录即可。想要重新实跑时，把缓存文件挪开并设置好 key。修改 TYPESAFE_MODEL 时也需要这样做，因为下面的缓存键只包含 state（上下文状态）和问题，而不包括模型。
以下数据来自 jev-1.13.0，采集日期为 2026-09-25。
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.13.0"
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")加载一段聊天
这 29 条消息是我们为一场 Elden Ring 直播亲自编写的，因此每种消息类型至少出现两次。其中两条是纯表情消息（emote 即 Twitch 上的小型聊天图片，输入时以 PogChamp 这种单词形式表示）。
一条消息究竟算不算剧透，或算不算主动支招（未经主播索求的游戏攻略），取决于当前在玩的是哪一款游戏。所以每条请求都要带上直播标题与游戏名称。
STREAM = {"title": "Elden Ring first playthrough! No spoilers please", "game": "Elden Ring"}
CHAT = [
("NightOwl_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 com"),
("LoreHound", "wait till you find out Ranni planned the Night of Black Knives"),
("kappa_kev", "LUL LUL LUL"),
("sunny_d", "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?"),
("LoreHound", "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_d", "who is the final boss?"),
("ggwp_andy", "LETS GOOOO"),
("NightOwl_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"),
("NightOwl_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]}})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"}}编写问题
我们会针对同一份 state（直播信息加上单条消息），在一次请求中提出三个问题，并一起作答。
category问题是一个Choice（选择题），覆盖五种消息类型。它所选中的标签（label）决定了该消息的处理方式：显示、高亮、变暗或隐藏。toxic问题是一个Noul（是非题）。若其概率（probability）达到 0.7 或更高（即下方设置的阈值TOXIC_AT），无论属于哪一类别，消息都会被隐藏。spoiler问题是一个Noul（是非题）。若其概率达到 0.7 或更高（即阈值SPOILER_AT），消息会被模糊处理，除非它已经被隐藏。
category 描述的是消息的主要意图。重复的表情符号和欢呼归入 hype；未经请求的推广归入 spam。仅含一个链接并不算 spam。若一条消息不属于其他任何类别，则归入 conversation 作为兜底。
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?"
),
}对单条消息进行分类
下面的 classify() 函数根据一条消息构建上下文状态（state）并发起请求。缓存以 state 和 questions 派生出一个键，把每个答案写入缓存，写成 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),
"input_tokens": response.usage.input_tokens,
}
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("LoreHound", "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.02
spoiler 0.09
LoreHound: wait till you find out Ranni planned the Night of Black Knives
conversation 0.57 #################
backseat 0.42 #############
hype 0.01
spam 0.00
question 0.00
toxic 0.03
spoiler 0.94
cozy_cat: you're so bad at this boss lmao, love you tho
conversation 0.81 ########################
hype 0.18 #####
backseat 0.01
question 0.00
spam 0.00
toxic 0.21
spoiler 0.02backseat 建议被归为 backseat，概率为 1.00，且 toxic 与 spoiler 都较低。关于「Night of Black Knives」的那条消息以微弱优势被归为 conversation 而不是 backseat，同时被 spoiler 这一题命中。最后一条消息看起来像是在骂人，但 toxic 概率仍然很低；由于结尾带了「love you tho」，它符合问题中「轻度调侃」的描述。
决定观众看到什么
下面的 decide() 函数只读取答案，不发起请求。它会检查三项内容，并按一组规则决定如何处理：
- 达到或超过
TOXIC_AT阈值的消息会被隐藏。 - category 对应规则为
hide的消息会被隐藏。 - 达到或超过
SPOILER_AT阈值的消息会被打码。
剩余的消息则按其 category 对应的规则处理。
在示例消息中，每条 toxic 概率要么不低于 0.97，要么不高于 0.24；每条 spoiler 概率要么不低于 0.94，要么不高于 0.20。因此在这两组数值之间任选一个阈值，结果都一样。你自己频道的聊天概率会落在中间区间；在实际使用之前，请把两个阈值都对照它跑一遍再上线。
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过滤整个聊天
上面的三条消息会被缓存起来，首次实际运行时还剩 26 个新请求。消息之间没有相互依赖，因此请求以八个为一组并发执行，pool.map 会按聊天顺序返回答案。
下面的输出就是观众看到的样子：被隐藏的消息只显示隐藏原因，被打码的消息只显示作者，高亮的问题用 >> 标记。每行右侧的标签就是 category 答案。
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}") NightOwl_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
LoreHound: ░░░░░░░░ spoiler conversation
kappa_kev: LUL LUL LUL hype
>> sunny_d: 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
LoreHound: ░░░░░░░░ 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_d: who is the final boss? question
ggwp_andy: LETS GOOOO hype
NightOwl_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
NightOwl_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整个过滤器就是 QUESTIONS、classify() 和 decide() 这三部分。要在自己的聊天数据上运行，只需替换 STREAM 和 CHAT，或者针对每个频道给 classify() 传入不同的 stream。
如果遇到限流，把 max_workers 调小后再运行；已经完成的答案都已缓存。这里 pool.map() 会保持聊天顺序，所以每个结果都按列表中的位置与对应的消息一一对应。在实际集成中，你也可以改为把平台的消息 ID 或消息页面元素的引用与每个请求绑定在一起，这样在答案返回时就能更新到正确的消息上。
不发起新请求地修改规则
答案描述了消息，decide() 把它们转换为动作。下面的示例隐藏 hype（欢呼和纯表情行）并弱化对话，复用上一步的 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")NightOwl_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)
NightOwl_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
NightOwl_22: this boss took me 40 tries lol show -> dim
14 messages changed, 0 new requests在直播频道使用前
- 每条消息都要等待它的答案。 本 cookbook 中请求的中位数约为 0.4 秒，因此新行出现在聊天中时，它的答案还没回来。要么先把每条新行隐藏起来等答案返回，要么接受一条不良消息会在屏幕上停留那么久。
- 单次请求跟不上繁忙聊天的节奏。 同时发起多个请求，例如 Filter the whole chat 中的八个。
- 本 cookbook 发布时，Jev 主要用英语训练。 它能接受其他语言，但对它们的回答准确度较低，因此非英语频道上的漏检会更多。
- 获胜类别仍可能不确定。 本示例始终套用获胜标签的规则。在上线直播前，请决定如何处理接近的类别概率和接近 0.5 的 Noul 值。可以把这类消息标记为待复核，或者保持可见而不是自动隐藏。在你自己频道的聊天里测试该策略。
- 剧透检查没有主播进度的记录。 它根据游戏标题和模型的知识来识别可能的剧情透露，而不是判断该观众或主播是否已经知道。要仅过滤尚未看到的剧情点，请提供当前进度和相关剧情信息，并调整 question 来使用这些上下文。
在 playground 中打开它
下面的链接会用场外指挥消息的 state 和填好的三个 question 打开 playground。在那里实时运行该调用，或编辑 question 后再次运行。
playground_link = make_playground_link(
{"stream": STREAM, "message": {"user": CHAT[4][0], "text": CHAT[4][1]}},
QUESTIONS,
models=[TYPESAFE_MODEL],
)
# Print a usable URL in a script; render a clickable link in a notebook.
if "__file__" in globals():
print(f"Open the state + questions in the TypeSafe playground:\n{playground_link}")
else:
display(Markdown(f"🔗 [Open the state + questions in the TypeSafe playground]({playground_link})"))在 TypeSafe playground 中打开 state + questions →
本文档由 Mintlify 构建并托管，它是一个开发者文档平台。