十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

从零构建 Twitter 表情回复机器人:API 接入、关键词匹配与部署实践

从零构建 Twitter 表情回复机器人:API 接入、关键词匹配与部署实践 我试过给直播群做一个自动回复表情的机器人最初是挂在 Discord 上后来发现很多朋友还是在 Twitter 上活跃得最多索性把它搬到 Twitter 上。这篇文章不是介绍那种群发营销号也不是什么复杂的人工智能而是一个非常具体的东西一个能听懂情绪关键词、然后在推文评论区自动回一张表情图的机器人。听起来好像不难但真正从零跑通里面涉及的开发者账号申请、接口调用、媒体上传、频率限制和误判处理坑远比想象中多。我这边把完整跑通的过程、核心代码、以及那些文档里没写清楚的细节都整理出来给也想做类似 Twitter 机器人不管是表情机器人、关键词提醒机器人、还是自动回复机器人的朋友一个可以直接参考的模板。1. 项目概述与核心需求解析1.1 表情机器人到底在做什么“Twitter 表情机器人”这个名字听起来有点玄拆开看就是一个很典型的“触发-匹配-回复”程序。它做的事情可以概括成三句话实时监听 Twitter 上符合条件的推文从推文文本里识别有没有命中预设的情绪关键词如果命中了就在这条推文下自动回复一张事先准备好的表情图也就是表情包。实际使用场景非常多。我最初的需求来自直播社群主播在 Twitter 上发一条“今天又被队友气到了”粉丝群里大家会默认刷一张“气到变形”的表情包。既然这个动作这么高频那就用机器人自动来做。再延伸一下品牌号也可以用它做趣味互动比如用户发推提到产品名机器人就在评论区回一个品牌专属表情学习打卡群还能做“你发‘打卡成功’我回你一张徽章图”的自动化贴纸玩法。这个机器人和市面上那些批量转发、自动点赞的“营销机器人”有本质区别它不依赖第三方营销平台没有爬取用户隐私表情库完全由你自己维护回复规则也完全透明。它只做一件事看到关键词回一张图然后把决定权交给你的规则。1.2 技术选型为什么是 Python Tweepy我一开始调研过三种方案分别说一下为什么最后选了 Python Tweepy。第一种是直接用 requests 请求 Twitter 的网页接口或非官方接口。这东西在早期还能用但现在平台的风控越来越严网页版接口里加了很多加密参数还需要处理登录态、验证码、限流算法而且一旦被识别账号风险很高。自己做着玩可以想稳定跑完全没必要。第二种是 Node.js 生态。以前有个 twit 库很流行但已经很久不维护了对 API v2 的支持很差很多接口要靠自己封装。而且媒体上传、过滤流这些功能在 Node.js 下还得自己写不少胶水代码。第三种就是 Python Tweepy。Tweepy 是 Twitter 官方 API 的第三方封装库对 API v2 支持非常完整包括流式过滤、推文发布、媒体上传都有现成方法。它的 API 设计也比较符合直觉一个项目从零到跑通代码量可以控制在两三百行内。而且文档和社区示例多遇到问题搜一下基本都有答案。当然选 Python 还有一个现实原因表情库的匹配逻辑后面大概率要加分词、正则、情感判断这些文本处理能力Python 在这些方面太方便了不需要再引一套技术栈。1.3 整体流程拆解整个机器人的运行流程可以拆成五步通过 Twitter API 拿到实时推文这一步可以用“过滤流”或“轮询”实现后面会详细说。提取推文文本做清洗和归一化比如统一大小写、去掉多余空格。把文本和表情库里的关键词规则做匹配命中后得到表情图路径。把表情图通过 Twitter Media 接口上传拿到 media_id。用这个 media_id 创建一条回复推文回复到原始推文下面。这五步看着简单实际操作时每一步都有细节第一步要解决权限和重连第二步要解决否定词和同义词第三步要处理多关键词冲突第四步要处理图片格式和大小限制第五步要处理重复回复。后面的内容基本就围绕这些细节展开。2. 环境准备与开发者账号配置2.1 申请 Twitter Developer 账号和 API Key写代码之前先把账号准备好。你需要一个 Twitter 账号然后去开发者平台创建一个应用。申请的时候会让你选用途我选的是“机器人自动回复”用途说明里写清楚“根据用户主动提及触发表情回复非垃圾信息”审核一般没问题。创建应用之后你能拿到几组重要凭证API Key / API Secret相当于应用的用户名和密码。Access Token / Access Token Secret代表应用以你的账号身份去操作回复推文用的就是这个身份。Bearer Token用于 API v2 的只读访问过滤流和读取推文都靠它。如果你的应用权限没有开启“读、写和消息”后面调用发布接口时会遇到 Forbidden 或 You are not allowed to access this endpoint 这类报错。所以创建应用后一定要去 User authentication settings 里把权限设为 Read and write并且把回调地址设置成 http://127.0.0.1:3000本地测试用。很多人会忽略的一个点Access Token 在申请页面生成的时候版本可能还是 v1 的格式。Tweepy 里传 token 时要注意用对顺序尤其是同时使用 API v1.1 的媒体上传和 API v2 的推文发布时两个客户端都要初始化token 不能混。2.2 本地开发环境搭建代码方面我用的是 Python 3.10你用 3.8 问题也不大。我们按下面的方式初始化项目mkdir twitter-emote-bot cd twitter-emote-bot python3 -m venv venv source venv/bin/activate pip install tweepy python-dotenvtweepy 负责调用 Twitter APIpython-dotenv 用来读取配置文件。我在项目根目录放一个 .env 文件API_KEY你的APIKey API_SECRET你的APISecret ACCESS_TOKEN你的AccessToken ACCESS_TOKEN_SECRET你的AccessTokenSecret BEARER_TOKEN你的BearerToken然后在代码里这样加载from dotenv import load_dotenv import os load_dotenv() consumer_key os.getenv(API_KEY) consumer_secret os.getenv(API_SECRET) access_token os.getenv(ACCESS_TOKEN) access_token_secret os.getenv(ACCESS_TOKEN_SECRET) bearer_token os.getenv(BEARER_TOKEN)这里有个小经验.env 文件不要提交到 git 仓库。我见过有人直接把密钥推到公开仓库里几分钟后就会被别人拿来刷接口。建议 .gitignore 至少加上 .env、venv/、pycache/、emotes/ 这些目录。2.3 验证凭证是否正确环境搭建完之后先不要急着写机器人先跑一个最小脚本确认凭证没问题import tweepy client tweepy.Client( bearer_tokenbearer_token, consumer_keyconsumer_key, consumer_secretconsumer_secret, access_tokenaccess_token, access_token_secretaccess_token_secret, ) me client.get_me() print(me.data.id, me.data.username)能打印出自己的用户 ID 和用户名说明凭证和网络链路都通了。这一步能过滤掉大量后面的“为什么发布不了”的问题。如果这里就报错多半是 Bearer Token 复制不全或者 Access Token 对应的是另一个账号重新生成一次就好。3. 机器人核心逻辑设计与实现3.1 表情库设计从字典到文件夹组织表情库是机器人的灵魂。最简单的方式是用一个 Python 字典来映射关键词和表情图路径EMOTE_MAP { 开心: emotes/happy.gif, 难过: emotes/sad.png, 愤怒: emotes/angry.gif, 抽卡: emotes/gacha.jpg, 打卡成功: emotes/checkin.gif, }这个方式适合验证逻辑但一旦表情多了维护起来痛苦。我建议做成“文件夹 JSON 索引”的结构emotes/ happy/ 开心.png 很开心.gif sad/ 难过.png 委屈.gif angry/ 愤怒.gif index.jsonindex.json 里存关键词到文件路径的映射加载逻辑统一走 JSON这样以后想加一个表情不用改代码只改配置和数据甚至可以把 index.json 丢给不懂代码的运营同事去维护。3.2 关键词匹配从“包含”到“带情感权重”匹配逻辑是最容易出 bug 的地方。如果只用“关键词 in 推文文本”这种包含判断很快会发现几个问题“不开心”会被“开心”命中但语义完全相反。“开心到变形”会被“开心”命中虽然也算沾边但不够精准。“今天真开心啊但是明天要加班”会同时命中“开心”和“加班”到底回哪张图大小写、繁体简体、全角半角都会影响匹配结果。我的做法是分三步处理。第一步文本清洗。把全角符号转半角去掉多余空格统一转小写import unicodedata def clean_text(text): text unicodedata.normalize(NFKC, text).lower() return .join(text.split())NFKC 标准化会把全角字符转成半角也能处理一些特殊空格。对于中文转小写影响不大但对英文字母来说是必要的。第二步否定词处理。我维护一个否定词表比如“不”“没”“莫”“别”。如果关键词前面近距离出现否定词就把这个关键词的权重设为负数NEGATIVE_WORDS [不, 没, 莫, 别, 无, 非] def has_negation(text, keyword, window5): index text.find(keyword) if index -1: return False start max(0, index - window) prefix text[start:index] return any(neg in prefix for neg in NEGATIVE_WORDS)这个逻辑简单但有效。窗口值我一般设成 5 个字太长容易把前面分句里的“不”也算进去太短又会漏掉“一点也开心不起来”。第三步多关键词冲突时按权重打分。每个表情规则可以配置一个权重命中多个规则时取分最高的回复。最初用字典就够了后续想精细控制再迭代成规则表。3.3 流式监听用过滤流拿到实时推文拿到推文的方式有两种轮询和流式。轮询实现简单就是定时去调 mentions 接口但延迟高而且很容易触发频率限制。过滤流是平台实时推给我们的方式只要推文满足你设置的规则API 会立刻把内容推过来延迟基本在秒级。我给这个机器人设置的过滤规则是推文必须包含机器人自己的 handle比如 emotebot。这样能避免机器人对全网所有推文都做匹配既省流量也让触发变得可控。用户只有主动 机器人机器人才会去回复。用 Tweepy 实现过滤流代码是这样的import tweepy class EmoteStream(tweepy.StreamingClient): def on_tweet(self, tweet): print(f收到推文: {tweet.text}) handle_tweet(tweet) stream EmoteStream(bearer_token) existing stream.get_rules() if existing.data: rule_ids [r.id for r in existing.data] stream.delete_rules(rule_ids) stream.add_rules(tweepy.StreamRule(valueemotebot)) stream.filter( tweet_fields[author_id, created_at], expansions[author_id], )第一次跑的时候如果之前已经添加过同名规则会报重复错误。我的习惯是先清空所有规则再添加。这里有个很隐蔽的坑过滤流拿到的推文可能包含自己的回复推文机器人回复后自己也会收到这条“新推文”如果不处理就会形成“机器人回复之后又收到回复消息再回复一次”的死循环。解决方法是在 handle_tweet 开头先判断推文作者是不是自己如果是就跳过。3.4 图片上传与回复推文匹配到表情图之后先要上传到 Twitter 媒体库拿到 media_id再用它创建回复推文。注意上传媒体走的是 API v1.1 的接口而创建推文走的是 API v2所以需要同时初始化两个客户端auth tweepy.OAuth1UserHandler( consumer_key, consumer_secret, access_token, access_token_secret ) api_v1 tweepy.API(auth, wait_on_rate_limitTrue) client_v2 tweepy.Client( bearer_tokenbearer_token, consumer_keyconsumer_key, consumer_secretconsumer_secret, access_tokenaccess_token, access_token_secretaccess_token_secret, ) def reply_with_emote(tweet_id, emote_path): media api_v1.media_upload(filenameemote_path) media_id media.media_id_string client_v2.create_tweet( in_reply_to_tweet_idtweet_id, media_ids[media_id], )关于媒体文件有几个限制要记牢静态图片最大 5MB常见格式 JPG、PNG、WEBP 都可以。GIF 动图最大 15MB。视频最大 512MB但会消耗额外的媒体额度表情机器人一般用不上。动图上传后可能会被平台转码画质会有轻微压缩属正常现象。如果上传后报“媒体处理超时”或者“格式不支持”优先检查是不是 GIF 帧数太多或尺寸异常。我用过的表情包里最容易出问题的就是那种几十 MB 的高清动图压到 10MB 以内基本就没问题了。3.5 防止重复回复流式连接断开重连后平台可能把断开期间的部分推文重新推一遍这时候机器人会对同一条推文回复两次。我在项目里用一个内存集合记录最近处理过的推文 IDseen_tweets set() MAX_RECORD 1000 def handle_tweet(tweet): if tweet.id in seen_tweets: return seen_tweets.add(tweet.id) if len(seen_tweets) MAX_RECORD: seen_tweets.clear() # 后续处理这个方案只适合单机进程重启后集合就空了。如果需要更可靠的去重可以把 tweet.id 存到 SQLite 或 Redis 里甚至直接用一个本地文件追加记录。对我的使用量来说内存集合已经够用但我会在系统里加一条日志方便事后追查“为什么重复回复了”。4. 部署上线与高频问题排查4.1 部署到云服务器的完整步骤本地跑通之后你不可能一直开着电脑。部署到云服务器上是必须的我用的是 systemd 做进程守护简单可靠。第一步把项目代码上传到服务器假定放在 /opt/twitter-emote-bot。第二步创建 systemd 服务文件 /etc/systemd/system/emote-bot.service[Unit] DescriptionTwitter Emote Robot Afternetwork.target [Service] Userubuntu WorkingDirectory/opt/twitter-emote-bot ExecStart/opt/twitter-emote-bot/venv/bin/python main.py Restartalways RestartSec10 EnvironmentFile/opt/twitter-emote-bot/.env [Install] WantedBymulti-user.target这里有个细节用 EnvironmentFile 加载 .env那么代码里 load_dotenv() 可以去掉避免两套配置源冲突。系统环境变量会被 os.getenv 直接读到。启动服务sudo systemctl daemon-reload sudo systemctl enable --now emote-bot sudo systemctl status emote-bot用 journalctl 看日志journalctl -u emote-bot -f日志里能直接看到“收到推文”和“回复成功”的记录。我习惯在关键节点加 print 并带时间戳排查问题会快很多。4.2 高频报错与排查速查表报错信息常见原因处理办法403 Forbidden应用权限没开 Read and write去开发者后台改权限改完需要等几分钟生效401 UnauthorizedAPI Key 或 Token 复制错误重新生成并核对注意区分 Access Token 和 Bearer Token429 Too Many Requests频率超限降低轮询频率如果是流式连接等待背压重试rule already exists添加了重复的过滤规则先 get_rules 再 delete_rules406 Not Acceptable过滤规则格式错误规则 value 尽量用 用户名别写太复杂media processing failed图片格式或大小超出限制压缩图片GIF 控制在 15MB 以内除了表里这些还有一个最容易忽略的问题应用刚创建时开发者权限可能还没有完全同步。如果你确信代码没问题但接口一直报错先把账号和应用退出重新登录一次很多时候是权限缓存的问题。4.3 频率限制与配额管理Twitter API 的频率限制是按“每个用户/每15分钟”计算的。过滤流本身没有严格的请求限制但轮询接口、媒体上传和推文发布都有配额。我的经验是把配额守则写进代码逻辑里发布回复前先检查剩余配额不够就直接跳过不硬试。流式断线重连时不要立刻重连等 10 到 30 秒给服务端留个缓冲。媒体上传尽量轻量化能提前压缩就提前压缩别把几 MB 的高清图反复传。配额管理看起来不起眼但它决定了机器人能不能长期稳定跑。我有一次把轮询间隔改成 10 秒跑了半天就触发 429整个账号的接口被临时限制连正常发推都受影响。后来改成流式 每次发布后 sleep(1)再也没遇到过这个问题。4.4 日志与监控让机器人自己“说话”机器人上线后最头疼的问题不是跑不起来而是不知道它为什么某天不回复了。我的做法是加两层监控。第一层是日志。代码里每个关键节点都输出一行日志格式统一为“时间、动作、结果”。比如2025-01-12 14:03:22 收到推文 189999999999 内容: emotebot 今天太开心了 2025-01-12 14:03:23 命中规则 happy - emotes/happy.gif 2025-01-12 14:03:24 上传媒体成功 media_id123456789 2025-01-12 14:03:25 回复成功 tweet_id190000000000第二层是心跳。我让机器人每隔 30 分钟写一条“我还活着”的标记到日志文件再用 crontab 定期检查日志文件的更新时间如果超过 40 分钟没更新就发一封邮件告警。对小项目来说这比上监控系统轻量得多。5. 真实运行体会与后续可以扩展的方向5.1 我跑这三周踩出的经验机器人在我这边稳定跑了三周整体感觉是核心逻辑并不难难的是内容规则和数据维护。表情库虽然只有几十个表情但维护成本比想象中高。你总会遇到“用户发了一条全新的表达机器没匹配上”的场景这时候你不能只加一个表情而是要去分析用户的习惯表达反向补全关键词表。另一个体会是关键词匹配在真实语料里远没有想象的干净。繁体、简体、全角半角、加空格、中间穿插字母各种变形都可能导致漏匹配。我后来加了一道清洗流程把所有文本先转成统一的简体形式匹配率明显提高了。如果让我重做这个项目我会在一开始就给表情库加上“规则版本”和“生效时间”两个字段这样后期迭代规则时能清楚知道哪些规则是旧的哪些是新加的修改起来也更有信心。5.2 可以继续玩的方向机器人的基础框架跑通后后面能扩展的方向很多接入 Discord 或 Telegram把同一个表情库复用过去一次维护多端发布。把关键词匹配从“规则表”升级成“关键词 情感判断”用简单的情感分析模型判断推文是积极还是消极再回对应的表情。做一个简单的后台页面直接在网页上编辑表情库和规则不用每次改代码。加上统计报表看哪些表情被触发最多哪些关键词从来没被用过方便优化表情库。这些扩展不需要动核心架构在你搭建的监听、匹配、回复框架上增量添加就行。最后再分享一个个人建议不要太早追求复杂的匹配算法。先把“看关键词回表情”这条主链路跑通把账号和接口的稳定性摸透再慢慢迭代规则。机器人这种东西稳定比聪明更重要。我在上线第一天因为回复过快被临时限制了十五分钟那一刻我才真正理解“克制请求频率”不是一句空话。希望这篇文章能帮你少走这几步弯路。
返回列表