背景说明:先把自动对话做轻,再考虑扩展功能
Telegram 机器人没有一个可以直接开启的“自动对话”按钮。要让机器人收到消息后自动回复,需要通过机器人平台或代码处理消息。本文采用 Python、aiogram 3 和 FastAPI 搭建一个轻量示例:用户询问价格、物流时返回固定答案;用户发送“预约”后,机器人接收下一条消息并确认预约信息。
这个方案适合咨询量不大、问题类型相对固定的店铺或社群。性能上的重点不是让机器人“记住所有聊天”,而是只保存当前对话步骤,设置过期时间,并通过 Webhook 接收消息。这样可以避免持续轮询、保存完整聊天记录,以及让简单问题也触发复杂的 AI 调用。下面的例子以小型网店为场景,你可以按自己的业务替换回复内容。
第一步:创建机器人并准备运行环境
在 Telegram 中找到官方账号 @BotFather,发送 /newbot,依照提示设置名称和用户名。创建完成后,BotFather 会提供一段 Bot Token。它相当于机器人的访问凭证,不能粘贴到公开网页、群聊或代码仓库中。
准备一台安装了 Python 3.10 或更高版本的服务器。创建项目目录后安装依赖:pip install aiogram fastapi uvicorn cachetools。将 Token 和 Webhook 密钥设置为环境变量,而不是直接写进代码。例如在 Linux 终端中执行:export BOT_TOKEN='替换为BotFather提供的Token',再设置 WEBHOOK_SECRET='自定义一段字母数字与下划线组成的密钥'。生产环境建议用服务器的环境变量管理功能保存这些值。
第二步:编写关键词回复和限时多轮状态
在项目目录新建 main.py,加入以下代码。示例只保存聊天 ID 对应的当前步骤,最多缓存 5000 条状态,并在 30 分钟后自动过期;不会把用户全部聊天内容长期留在内存里。
import os
from cachetools import TTLCache
from fastapi import FastAPI, Request, HTTPException
from aiogram import Bot, Dispatcher
from aiogram.types import Update
TOKEN = os.environ["BOT_TOKEN"]
SECRET = os.environ["WEBHOOK_SECRET"]
bot = Bot(token=TOKEN)
dp = Dispatcher()
app = FastAPI()
states = TTLCache(maxsize=5000, ttl=1800)
@dp.message()
async def reply(message):
text = (message.text or "").strip()
chat_id = message.chat.id
if not text:
return
if states.get(chat_id) == "need_order":
states.pop(chat_id, None)
await message.answer("收到,我们已记录:" + text + "。客服会尽快确认。")
return
if text.lower() in ("预约", "联系人工"):
states[chat_id] = "need_order"
await message.answer("请发送商品名称和方便联系的时间,不需要提供银行卡或密码。")
elif "价格" in text or "多少钱" in text:
await message.answer("本店常规款 99 元起,具体价格请发送商品名称。")
elif "物流" in text or "快递" in text:
await message.answer("订单发出后可凭订单号查询物流;需要人工协助请发送“预约”。")
else:
await message.answer("我可以查询价格、物流或登记预约,请发送对应关键词。")
@app.post("/telegram")
async def telegram_webhook(request: Request):
if request.headers.get("X-Telegram-Bot-Api-Secret-Token") != SECRET:
raise HTTPException(status_code=403, detail="Forbidden")
data = await request.json()
update = Update.model_validate(data, context={"bot": bot})
await dp.feed_update(bot, update)
return {"ok": True}代码里的关键词和回复是示例,可替换成实际业务内容。为了让规则匹配更稳定,建议先处理最常见、答案明确的问题;不要把一整段聊天历史放入状态缓存。若用户的下一条消息可能包含敏感信息,也要在提示语中明确要求不要发送密码、验证码或支付信息。
第三步:启动服务并配置 HTTPS Webhook
Webhook 需要一个 Telegram 能访问的公网 HTTPS 地址。可以将域名通过 Nginx 等反向代理转发到本机的 8000 端口,并配置有效的 TLS 证书。启动应用:uvicorn main:app --host 0.0.0.0 --port 8000。本机服务启动并不代表 Telegram 已经能访问它;如果没有公网域名或 HTTPS,可以先在测试环境使用安全的隧道服务,但不要把 Token 暴露在浏览器页面或日志中。
服务可从公网访问后,调用 Telegram 的 setWebhook 接口登记地址。将命令中的 TOKEN、域名和密钥替换成自己的值:curl -F "url=https://你的域名/telegram" -F "secret_token=你的WEBHOOK_SECRET" https://api.telegram.org/bot你的TOKEN/setWebhook。正常情况下会返回 "ok":true。随后在 Telegram 中向机器人发送“价格”或“预约”,检查是否收到对应回复。不要把尖括号占位符原样复制到命令里。
第四步:用真实消息检查资源与回复效果
测试时至少覆盖三种情况:常见问题能否直接命中回复;发送“预约”后,紧接着发送的内容能否触发确认;发送其他文本时,机器人能否给出清楚的下一步提示。再重启服务测试一次,确认自己理解当前状态存放在内存中的影响:服务重启后,未完成的预约步骤会消失。如果业务要求对话状态在重启后保留,应改用带过期时间的 Redis 等外部缓存,而不是无限期把所有聊天内容保存在进程内存。
可以同时查看服务器 CPU、内存和错误日志。如果普通关键词回复也造成明显负载,先检查是否重复启动了多个轮询进程、是否在每条消息中读取大型文件,或是否无条件调用耗时的外部服务。对价格、营业时间等固定问答,直接查字典或使用明确的条件判断,通常比每条消息都请求 AI 接口更省资源,也更容易保证答案一致。
注意事项:避免重复处理与缓存膨胀
- 不要同时对同一个机器人运行长轮询程序和 Webhook 接收程序。切换接收方式时应检查旧进程或旧部署,避免收不到消息或重复处理。
- Webhook 地址必须使用有效 HTTPS;核对 setWebhook 返回结果,并使用密钥头校验请求,避免不相关的请求触发机器人逻辑。
- 示例状态只存在当前 Python 进程中,多进程部署时各进程的缓存互不共享。需要多实例或重启后保留状态时,使用 Redis 等共享存储,并为键设置 TTL。
- 不要无上限地保存消息、用户资料或请求日志。日志仅记录排错需要的信息,并避免记录 Token、验证码和用户敏感内容。
- 外部接口、图片处理和 AI 回复可能很慢。遇到这类任务时应设置超时与错误处理;流量较高时可考虑队列,避免请求处理被长任务拖住。
- 机器人只能回复它实际收到且有权限处理的消息;群组隐私设置、用户是否启动机器人等因素,也可能影响消息能否送达。
常见问题
问:不懂 Python,能不能添加自动对话?
答:可以使用支持 Telegram 触发器和条件分支的自动化平台配置关键词回复,但仍要确认平台的消息触发方式、运行配额、数据保存期限和隐私条款。本文代码适合希望控制服务器资源和回复规则的用户。
问:为什么发送消息后机器人没有回复?
答:依次检查服务是否运行、域名 HTTPS 是否可访问、setWebhook 是否成功、环境变量是否正确,以及服务日志中是否出现 403 或程序异常。也可以调用 Telegram 的 getWebhookInfo 接口查看 Webhook 状态与错误提示。
问:状态缓存设置 5000 条是不是代表只能有 5000 个用户?
答:不是严格的用户总数限制,而是最多缓存 5000 个尚未过期的对话状态。过期状态会被清理;如果咨询峰值超过预期,应观察内存使用,并结合实际并发调整上限或改用外部缓存。
问:可以让机器人记住完整对话并像真人一样持续聊天吗?
答:可以接入 AI 服务或数据库实现更复杂的上下文,但这会增加网络请求、响应延迟、存储成本和隐私风险。建议先明确保留哪些必要字段、保存多久以及如何删除,再按业务需求逐步增加功能,不要默认永久保存完整聊天记录。