简介这是一套基于Python与Pyrogram框架实现的人形机器人Telegram Bot完整开发资源面向计算机专业本科生及初级开发者适用于毕业设计、课程设计与中小型项目快速原型开发。资源包含58个文件主体为40个Python源码文件含主程序main.py、调度器schedulers、模型加载逻辑及消息响应模块辅以10个ONNX格式预训练模型文件支撑动作识别与行为推理、3个YAML配置文件定义机器人行为规则与启动参数以及README.md等文档类文件整体压缩包仅1.89MB轻量易部署。已有62人学习下载体现了其在教学实践场景中的实用价值。用户可直接运行调试获得完整的Telegram机器人交互流程、SQLite本地状态管理、ONNX模型轻量化推理集成方案以及清晰分层的代码结构app/、libs/、models/等模块划分明确便于理解Bot架构设计与AI能力嵌入逻辑。1. 这不是 Telegram Bot而是一个带动作逻辑、状态记忆和多模态响应能力的「人形机器人」服务端框架你用 Pyrogram 写过群管理 Bot发个消息、踢个人、定时转发——那只是 Telegram 协议的“遥控器”。但这个项目里main.py启动后监听的不是/start而是reply_message.yaml里定义的「情绪触发词」models/下放的不是.pth而是 ONNX 格式的姿态生成模型schedulers/里跑的不是 APScheduler 的 cron 任务而是基于用户活跃度动态调整的「拟人化响应节律」。它把 Telegram 会话当成了人机交互的物理接口用户说“我累了”机器人不回“收到”而是调用libs/emotion_engine.py计算当前对话轮次的情绪衰减系数再从onnxes/pose_generator.onnx推理出一个低头缓慢眨眼的 SVG 动作序列最后封装成reply_message.yaml中预设的富文本卡片发回去。适合课程设计的同学快速搭出有“行为逻辑”的 Bot 框架也适合毕业设计学生在config.py里替换自己的语音合成模块或视觉反馈组件——它不卖功能卖的是可拆解、可插拔、带状态机的人形交互骨架。2. 从零部署环境搭建、配置初始化与核心服务启动流程2.1 环境隔离与依赖安装为什么必须用 Python 3.9 而非系统默认版本Pyrogram 2.0.107 对asyncio的 event loop 处理有重大变更而本项目中schedulers/heartbeat_scheduler.py依赖asyncio.run_coroutine_threadsafe()在后台线程安全调度心跳任务。Python 3.8 及以下版本在某些 Linux 发行版如 Ubuntu 20.04上会因loop.is_running()判断异常导致调度器静默退出。实测 Python 3.9.18 是最稳定的基线版本。提示不要用sudo pip install全局安装。所有操作必须在虚拟环境中进行否则supervisord.conf启动时会因路径污染导致ImportError: cannot import name Client from pyrogram。# 创建隔离环境推荐使用 pyenv 或 conda此处以 venv 为例 python3.9 -m venv tgbot-env source tgbot-env/bin/activate # 安装核心依赖注意requirements.txt 中 pyrogram 版本已锁定为 2.0.107 pip install --upgrade pip pip install -r requirements.txt # 验证关键组件版本必须输出匹配值 python -c import pyrogram; print(pyrogram.__version__) # 输出应为2.0.107 python -c import onnxruntime; print(onnxruntime.__version__) # 输出应为1.16.0ONNX Runtime 必须 ≥1.15.0否则 pose_generator.onnx 加载失败上述命令执行后pyrogram和onnxruntime的版本必须严格匹配。若pip install -r requirements.txt报错onnxruntime-cuda冲突请手动卸载并重装 CPU 版本pip uninstall onnxruntime onnxruntime-gpu -y pip install onnxruntime1.16.0原因项目onnxes/目录下所有模型均为 CPU 推理优化格式无 CUDA op强行安装 GPU 版本会导致onnxruntime.InferenceSession初始化时抛出InvalidGraph异常且错误堆栈不提示具体节点问题极易误判为模型损坏。2.2 配置文件初始化从 config_example.py 到可运行 config.py 的三步校验项目不提供开箱即用的config.py必须手动复制并填写。这不是为了增加门槛而是规避 Telegram API Key 泄露风险——Git 历史中一旦提交真实api_id该 Key 就永久失效。cp config_example.py config.py nano config.py需修改的仅三个字段其余保持默认字段名填写要求验证方式API_IDTelegram App 的整数 ID从 https://my.telegram.org/apps 获取启动时报ValueError: api_id must be an integer即未填或填错类型API_HASH与 API_ID 对应的 32 位十六进制字符串启动时报Unauthorized: Invalid API key表示 hash 错误或 ID 不匹配BOT_TOKENBotFather 分配的xxx:yyy格式 Token启动时报Forbidden: bot token is invalid表示格式错误含空格/换行或已被撤销注意BOT_TOKEN必须是Bot 类型账号的 Token不能是 User Session。项目中login.py仅用于首次生成session.session文件后续运行完全依赖BOT_TOKEN。若误用 User TokenClient初始化时会卡在Connecting...状态超过 60 秒后超时。验证配置是否生效的最小测试命令python -c from config import API_ID, API_HASH, BOT_TOKEN assert isinstance(API_ID, int), API_ID must be integer assert len(API_HASH) 32, API_HASH must be 32-char hex string assert : in BOT_TOKEN and len(BOT_TOKEN.split(:)) 2, BOT_TOKEN format error print(✅ Config validation passed) 2.3 启动服务与会话初始化login.py的真实作用与session.session文件生成逻辑login.py并非“登录脚本”而是Telegram 用户会话凭证生成器。它只在首次部署时运行一次目的是生成app/session.session文件供main.py中的Client实例复用。其本质是调用 Pyrogram 的Client.start()方法触发 Telegram 的 SMS/验证码登录流程。python login.py执行后终端将提示Please enter your phone number (with country code, e.g. 8613800138000):输入手机号含国际区号Telegram 将发送验证码。关键点此过程必须使用真实手机号注册的 Telegram 账号且该账号不能是 BotBot 账号无法生成 session。生成的session.session是 SQLite 数据库文件存储了加密的 auth key 和设备信息。提示session.session文件生成后login.py可删除或保留。但若更换服务器或重装系统必须重新运行login.py—— 因为 session 绑定设备指纹跨设备复用会触发 Telegram 的“新设备登录”风控导致main.py启动时报SessionPasswordNeeded。main.py启动逻辑依赖此 session# main.py 片段 app Client( session, # session name → 自动读取 app/session.session api_idAPI_ID, api_hashAPI_HASH, bot_tokenBOT_TOKEN, workdirapp/ # 显式指定 session 存储路径 )若app/session.session不存在app.start()将抛出FileNotFoundError若存在但已过期如 Telegram 主动吊销则报AuthKeyUnregistered此时必须重新运行login.py。3. 核心模块解析人形行为引擎如何驱动 Telegram 消息流3.1reply_message.yaml结构化响应规则引擎的设计哲学与字段语义这不是简单的关键词-回复映射表而是一个带条件分支、状态上下文和动作指令的 DSL领域特定语言。YAML 文件被filters/reply_filter.py解析为RuleSet对象每个rule包含trigger,response,context,action四个一级键。# reply_message.yaml 示例片段 - trigger: text: [我饿了, 肚子咕咕叫] regex: .*[饿|馋].* response: text: 正在调用厨房机器人准备便当… media: onnxes/kitchen_pose.onnx # 指向 ONNX 模型路径 context: required_state: [user_hungry] timeout: 300 # 5分钟内有效 action: type: pose_generate params: {speed: 0.8, repeat: 1}关键字段说明trigger.text精确匹配数组区分大小写支持中文trigger.regex正则表达式用于模糊意图识别如匹配“好[累|困|乏]”response.media不是图片路径而是 ONNX 模型文件名由models/pose_generator.py加载并推理出 SVG 动作帧context.required_state检查app/state_manager.py中维护的用户状态栈防止“我饿了”在用户刚吃完饭后被重复响应action.type目前支持pose_generate调用姿态模型、audio_play播放 TTS 音频、delay_reply延迟响应模拟思考。注意media字段值必须与onnxes/目录下文件名完全一致含扩展名。若写成kitchen_pose而实际文件为kitchen_pose.onnxonnxruntime.InferenceSession将抛出FileNotFoundError且错误日志不提示缺失.onnx后缀。3.2models/pose_generator.pyONNX 模型加载、推理与 SVG 动作序列生成全流程人形机器人的“动作”本质是 SVG 路径动画。pose_generator.py将 ONNX 模型输出的 128 维向量解码为path dM10,20 L30,40 ...字符串并嵌入 Telegram 支持的InputMediaAnimation格式。# models/pose_generator.py 关键逻辑 import onnxruntime as ort import numpy as np class PoseGenerator: def __init__(self, model_path: str): self.session ort.InferenceSession(model_path) # 加载 ONNX 模型 self.input_name self.session.get_inputs()[0].name self.output_name self.session.get_outputs()[0].name def generate_svg(self, input_vector: np.ndarray) - str: # 输入(1, 128) 归一化向量输出SVG 字符串 result self.session.run([self.output_name], {self.input_name: input_vector}) svg_data self._decode_to_svg(result[0][0]) # 自定义解码函数 return svg_datainput_vector来自reply_message.yaml中action.params的 JSON 序列化结果如{speed: 0.8}→[0.8, 0, 0, ..., 0]填充至 128 维。_decode_to_svg()函数将模型输出的 64 个控制点坐标x,y转换为贝塞尔曲线路径。实测发现若input_vector维度不是(1, 128)ONNX Runtime 会静默返回全零数组导致 SVG 生成为空路径path dTelegram 发送时显示空白卡片——这是最隐蔽的“黑匣子”坑。验证模型可用性的最小脚本# test_pose.py import numpy as np from models.pose_generator import PoseGenerator pg PoseGenerator(onnxes/stand_pose.onnx) test_input np.random.rand(1, 128).astype(np.float32) # 必须 float32 svg pg.generate_svg(test_input) print(✅ SVG length:, len(svg)) print(✅ Contains path tag:, path in svg)3.3schedulers/heartbeat_scheduler.py基于用户活跃度的拟人化响应节律控制器人形机器人不能秒回每条消息——那像客服机器人。本项目通过HeartbeatScheduler实现「思考延迟」用户连续发送 3 条消息第 4 条开始启用random.uniform(1.2, 3.5)秒延迟若 10 分钟无交互则自动发送“我在待机中… ”并降低心跳频率。# schedulers/heartbeat_scheduler.py class HeartbeatScheduler: def __init__(self, app: Client): self.app app self.user_activity {} # {user_id: [last_timestamp, msg_count]} self.base_delay 0.8 # 基础响应延迟秒 async def adjust_delay(self, user_id: int) - float: now time.time() if user_id not in self.user_activity: self.user_activity[user_id] [now, 1] return self.base_delay last_ts, count self.user_activity[user_id] if now - last_ts 600: # 10分钟无交互 self.user_activity[user_id] [now, 1] await self.app.send_message(user_id, 我在待机中… ) return 5.0 # 待机模式延迟 # 活跃度计算5分钟内消息数越多延迟越长 if now - last_ts 300: count 1 self.user_activity[user_id] [now, count] return max(self.base_delay * (1.0 count * 0.3), 0.5) # 上限 3.0s else: self.user_activity[user_id] [now, 1] return self.base_delay该调度器在filters/message_filter.py中被调用# filters/message_filter.py Client.on_message(filters.private ~filters.command([start, help])) async def handle_private_message(client, message): delay await heartbeat_scheduler.adjust_delay(message.from_user.id) await asyncio.sleep(delay) # 真实延迟 await reply_to_message(client, message) # 执行响应逻辑提示asyncio.sleep()是唯一合法的延迟方式。若用time.sleep()整个事件循环将被阻塞导致其他用户消息积压——这是新手最常翻车的操作。4. 避坑指南部署与调试中高频出现的 5 类血泪问题4.1 现象main.py启动后立即退出日志仅显示INFO:root:Starting...无后续原因config.py中BOT_TOKEN格式错误如开头/结尾含空格、含不可见 Unicode 字符、或使用了 User Token 而非 Bot Token。Pyrogram 在Client.start()前会校验 Token 格式失败则静默退出。解决用cat -A config.py | grep BOT_TOKEN查看隐藏字符确保BOT_TOKEN是纯 ASCII 字符串且符合数字:字母数字混合格式如123456789:ABCdefGHIjklMNOpqrSTUvwxYZ123456789。4.2 现象发送消息后机器人无响应app.log中反复出现WARNING:pyrogram.session.session:Recovering...原因app/session.session文件损坏或权限不足。常见于用root运行login.py生成 session再用普通用户运行main.py导致 session 文件属主为 root普通用户无读取权限。解决chmod 600 app/session.session chown $USER:$USER app/session.session若仍报错删除app/session.session并重新运行login.py。4.3 现象reply_message.yaml中定义的media: walk_pose.onnx无法触发动作日志报FileNotFoundError: onnxes/walk_pose.onnx原因onnxes/目录下实际文件名为walk_pose.onnx.bin或walk_pose.onnx.txt下载时被浏览器重命名或media字段值与文件名大小写不一致Linux 系统区分大小写。解决ls -l onnxes/确认真实文件名grep -n walk_pose reply_message.yaml检查 YAML 中拼写确保media值与onnxes/下文件名逐字节相同。4.4 现象pose_generator.py报错onnxruntime.capi.onnxruntime_pybind11_state.InvalidArgument: [ONNXRuntimeError] : 2 : INVALID_ARGUMENT : Invalid shape for input原因ONNX 模型期望输入维度为(1, 128)但代码中传入了(128,)或(128, 1)。numpy.reshape()未正确处理 batch 维度。解决强制 reshape 为(1, 128)input_tensor np.array(params_list, dtypenp.float32).reshape(1, -1) # 确保 -1 展开后长度为 128否则报错 assert input_tensor.shape (1, 128), fExpected (1,128), got {input_tensor.shape}4.5 现象supervisord.conf启动后supervisorctl status显示FATAL日志提示ImportError: No module named pyrogram原因Supervisor 默认使用系统 Python 解释器而非项目虚拟环境中的 Python。supervisord.conf中未指定environment或command路径。解决修改supervisord.conf[program:tgbot] command/path/to/tgbot-env/bin/python main.py ; ← 必须写绝对路径 directory/path/to/project/root environmentPATH/path/to/tgbot-env/bin autostarttrue autorestarttrue5. 进阶技巧如何用fix_sqlite.py修复会话数据与自定义状态持久化5.1fix_sqlite.py的真实用途抢救被 Telegram 强制登出的 session 数据app/session.session是 SQLite 数据库但 Pyrogram 不直接暴露其 schema。当 Telegram 主动吊销 session如用户在手机端点击“终止所有会话”session.session中的auth_key表会被清空main.py启动时报AuthKeyUnregistered。此时fix_sqlite.py并非“修复”而是重建最小必要表结构让login.py能绕过旧 session 直接发起新登录。python fix_sqlite.py --rebuild该脚本执行以下操作删除app/session.session备份为session.session.bak创建新 SQLite 文件仅建sessions表Pyrogram 最小依赖插入空记录使Client.start()不因表缺失而崩溃。注意--rebuild会丢失所有历史消息缓存但不影响 Bot Token 功能。Bot 的消息收发完全独立于 session 文件只有 User 类型客户端才依赖 session。5.2 自定义状态持久化在app/state_manager.py中接入 Redis 替代内存存储默认state_manager.py使用dict存储用户状态进程重启即丢失。生产环境需持久化。以下是接入 Redis 的最小改造# app/state_manager.py import redis import json class StateManager: def __init__(self): # 原逻辑self.states {} self.redis_client redis.Redis(hostlocalhost, port6379, db0) def set_state(self, user_id: int, state: str, timeout: int 300): key fuser:{user_id}:state self.redis_client.setex(key, timeout, state) # 自动过期 def get_state(self, user_id: int) - str: key fuser:{user_id}:state state self.redis_client.get(key) return state.decode(utf-8) if state else None # 在 main.py 中替换实例 # from app.state_manager import StateManager # state_manager StateManager() # ← 替换原内存版Redis 配置要点必须安装redis-pypip install redis4.6.0兼容 Python 3.9redis.Redis连接参数需与实际 Redis 服务匹配db0是默认库setex的timeout单位为秒与reply_message.yaml中context.timeout单位一致。5.3 验证状态持久化的三步测试法写入验证发送/setstate hungry需在filters/command_filter.py中添加该命令检查 Redis 中是否存在对应 keyredis-cli KEYS user:*:state # 应返回类似 user:123456789:state redis-cli GET user:123456789:state # 应返回 hungry读取验证重启main.py后发送触发required_state: [hungry]的消息如“我饿了”观察是否正常响应而非跳过规则。过期验证等待timeout秒后再次发送同一消息确认响应变为默认 fallback如“没理解你的状态”证明 Redis TTL 生效。从那以后我每次部署新环境都强制走一遍python fix_sqlite.py --rebuild python login.py流程哪怕session.session看似存在——因为 Telegram 的 session 吊销机制从不通知客户端静默失效才是常态。而reply_message.yaml的每一次修改我必用python -c import yaml; print(yaml.safe_load(open(reply_message.yaml)))验证语法避免 YAML 缩进错误导致整条规则失效。希望帮到你。本文还有配套的精品资源点击获取