1. 项目缘起从“MoltBot”到“tele-code”的极简之路最近在折腾一些自动化工具发现很多现成的方案要么太重要么配置复杂得让人头疼。特别是想给Telegram做个能处理代码、执行简单任务的机器人一搜要么是各种依赖的庞然大物要么是年久失修的老古董。正好看到社区里有人在讨论一个叫“OpenClaw”的项目后来好像又改名叫“tele-code”了核心思路是用极简的代码实现一个功能性的Telegram机器人。这勾起了我的兴趣作为一个喜欢自己动手的开发者我决定顺着这个思路尝试用大约400行核心代码从头打造一个属于自己的、功能清晰的“极简版MoltBot”。这个项目的核心目标很明确用最少的代码实现一个能通过Telegram Bot API接收消息、理解用户意图特别是与代码/指令相关的、并调用本地或远程服务比如大语言模型、代码执行环境进行处理和回复的机器人框架。它不追求大而全而是聚焦于“接收-解析-执行-回复”这个核心链路确保代码清晰、易于理解和二次开发。网上流传的“OpenClaw”或“tele-code”更像是一个概念或者一个实现范本其精髓在于架构设计而非某个固定的代码库。因此我的实现过程实际上是对这个极简机器人架构思想的一次实践和拆解。为什么是400行这不是一个硬性限制而是一个设计导向。它强迫你在实现核心功能时必须做出取舍避免陷入过度设计的陷阱。每一个函数、每一个类都必须有其不可替代的价值。最终这个“玩具”不仅能帮你理解Telegram Bot的工作原理更能成为一个可扩展的基石你可以基于它快速添加新的技能Skill比如代码解释、命令执行、文件处理甚至是与本地部署的Ollama大模型对话。接下来我就把自己从零搭建这个极简机器人的过程、遇到的坑以及思考分享出来。2. 极简架构设计核心模块拆解与选型思考一个能用的Telegram机器人最核心的流程无非是长轮询或Webhook获取消息 - 解析消息内容 - 根据内容匹配并执行相应逻辑 - 构造回复并发送。我们的400行代码就要围绕这个流程展开。首先我们需要做出几个关键的技术选型。2.1 通信层长轮询 vs WebhookTelegram Bot API支持两种方式接收更新getUpdates长轮询和setWebhookWebhook。Webhook需要你有一个公网可访问的HTTPS端点这对于本地开发或内网环境不太友好。而getUpdates方式简单粗暴适合快速原型开发。我们的极简版显然选择后者。我们只需要一个循环不断地调用getUpdates接口获取新的消息。这里有一个关键参数offset用于确认已处理的消息避免重复处理。2.2 依赖与框架裸奔还是用SDK为了极致精简我决定不使用任何重量级的Telegram Bot SDK比如python-telegram-bot。这些SDK功能强大但也会引入大量我们暂时不需要的抽象和依赖。我们的目标是理解底层原理所以直接使用Python内置的requests库来调用Telegram Bot的HTTP API。这样整个项目对外部依赖的需求降到最低只需要requests即可。这符合“极简”的哲学。2.3 核心执行引擎技能Skill系统机器人的“智能”体现在它能做什么。我们设计一个简单的“技能”系统。每个技能Skill都是一个独立的处理单元负责判断自己是否能处理当前消息通过关键词、命令前缀等并能执行具体的任务。例如EchoSkill: 复读机技能匹配/echo命令回复用户发送的内容。CodeRunSkill: 代码执行技能匹配如“/run python print(‘hello’)”这样的消息在安全的沙箱中执行代码并返回结果。LLMSkill: 大模型对话技能匹配普通文本对话将问题转发给配置好的大模型如本地Ollama并返回回复。技能系统的优势在于解耦和可扩展。主程序只负责消息路由具体的功能由各个技能插件实现。新增功能时只需编写一个新的技能类并注册即可无需修改核心流程。2.4 配置与安全即使是极简版配置和安全也不能忽视。我们需要一个配置文件如config.yaml或.env来存放Bot Token、模型端点等敏感信息。绝对不要将Token硬编码在代码中对于代码执行这类危险操作必须考虑安全沙箱例如使用docker run限制代码的运行环境或者使用一些安全的代码执行库如piston_api避免执行任意代码对主机造成破坏。基于以上思考我画出了这个极简机器人的心智模型一个由Main Loop驱动通过Dispatcher将消息分发给注册的Skill最终由Sender回复的流水线。代码将围绕这几个核心模块展开。3. 核心代码逐行实现从零搭建流水线接下来我们进入具体的代码实现环节。我会分模块讲解并附上关键代码段和解释。假设我们的项目名为mini_moltbot。3.1 项目初始化与配置管理首先创建项目结构并使用pip install requests pyyaml安装最小依赖。然后创建config.yamltelegram: bot_token: “YOUR_BOT_TOKEN_HERE” # 从 BotFather 获取 api_url: “https://api.telegram.org/bot” skills: llm: enabled: true base_url: “http://localhost:11434” # Ollama 默认地址 model: “llama3.2:1b” # 使用的模型名称 code_runner: enabled: true safe_mode: “docker” # 或 “piston” docker_image: “python:3.9-slim”创建一个config.py来读取配置import yaml import os class Config: def __init__(self, config_path“config.yaml”): with open(config_path, ‘r’) as f: self.data yaml.safe_load(f) # 也可以支持环境变量覆盖 self.bot_token os.getenv(“BOT_TOKEN”, self.data[‘telegram’][‘bot_token’]) config Config()注意在实际部署时更推荐使用环境变量来传递BOT_TOKEN等机密信息避免配置文件泄露。3.2 通信核心Telegram API 封装我们创建一个telegram_client.py封装最基础的发送消息和获取更新的逻辑。import requests import time from config import config class TelegramClient: def __init__(self): self.token config.bot_token self.base_url f“{config.data[‘telegram’][‘api_url’]}{self.token}” self.last_update_id 0 # 用于记录已处理的最新update_id def get_updates(self, timeout30): “”“长轮询获取消息更新”“” params {‘timeout’: timeout, ‘offset’: self.last_update_id 1} try: resp requests.get(f“{self.base_url}/getUpdates”, paramsparams, timeouttimeout5) resp.raise_for_status() updates resp.json().get(‘result’, []) if updates: self.last_update_id updates[-1][‘update_id’] return updates except requests.exceptions.RequestException as e: print(f“获取更新失败: {e}”) return [] def send_message(self, chat_id, text, parse_modeNone): “”“发送文本消息”“” payload {‘chat_id’: chat_id, ‘text’: text} if parse_mode: payload[‘parse_mode’] parse_mode try: resp requests.post(f“{self.base_url}/sendMessage”, jsonpayload) resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: print(f“发送消息失败: {e}”) return None这个类非常精简只实现了我们最需要的两个方法。get_updates中的offset管理是关键它确保了消息不会丢失或重复。3.3 技能系统抽象与实现创建skill.py作为基类然后实现具体技能。from abc import ABC, abstractmethod class Skill(ABC): “”“技能基类”“” abstractmethod def can_handle(self, message_text: str) - bool: “”“判断是否能处理此消息”“” pass abstractmethod def handle(self, message_text: str, chat_id: int, **kwargs) - str: “”“处理消息并返回回复文本”“” pass接着实现一个简单的复读技能echo_skill.pyfrom skill import Skill import re class EchoSkill(Skill): def can_handle(self, message_text): # 匹配 /echo 开头的命令 return message_text.startswith(‘/echo’) def handle(self, message_text, chat_id, **kwargs): # 提取 /echo 后面的内容 content message_text[5:].strip() if not content: return “用法: /echo 你要我说的话” return f“你说: {content}”再实现一个稍微复杂点的llm_skill.py用于连接Ollamafrom skill import Skill import requests from config import config class LLMSkill(Skill): def __init__(self): self.base_url config.data[‘skills’][‘llm’][‘base_url’] self.model config.data[‘skills’][‘llm’][‘model’] self.enabled config.data[‘skills’][‘llm’][‘enabled’] def can_handle(self, message_text): # 这是一个默认技能当没有其他技能匹配时由它来处理普通对话 # 在实际路由中我们会最后才检查这个技能 return self.enabled and not message_text.startswith(‘/’) def handle(self, message_text, chat_id, **kwargs): if not self.enabled: return “大模型技能未启用。” try: resp requests.post( f“{self.base_url}/api/generate”, json{ “model”: self.model, “prompt”: message_text, “stream”: False }, timeout60 ) resp.raise_for_status() reply resp.json().get(‘response’, ‘模型没有返回内容。’) return reply except requests.exceptions.ConnectionError: return “无法连接到大模型服务请检查Ollama是否运行在 {self.base_url}。” except Exception as e: return f“调用大模型时出错: {str(e)}”3.4 消息分发器大脑中枢创建dispatcher.py它是连接所有部分的核心。class Dispatcher: def __init__(self): self.skills [] def register_skill(self, skill): “”“注册一个技能”“” self.skills.append(skill) def dispatch(self, message_text, chat_id): “”“将消息分发给第一个能处理的技能”“” for skill in self.skills: if skill.can_handle(message_text): return skill.handle(message_text, chat_id) # 如果没有技能匹配返回默认提示 return “抱歉我还没学会处理这个。你可以试试 /echo 或直接和我聊天如果配置了LLM。”分发器的逻辑是顺序匹配。注册技能的顺序很重要通常把匹配规则明确的命令技能如/echo放在前面把兜底的技能如LLMSkill放在最后。3.5 主循环让机器人跑起来最后在main.py中我们把所有模块组装起来。from telegram_client import TelegramClient from dispatcher import Dispatcher from echo_skill import EchoSkill from llm_skill import LLMSkill import time def main(): # 初始化客户端和分发器 client TelegramClient() dispatcher Dispatcher() # 注册技能 dispatcher.register_skill(EchoSkill()) dispatcher.register_skill(LLMSkill()) # LLM技能放在最后作为兜底 print(“Mini MoltBot 启动...“) while True: updates client.get_updates() for update in updates: # 提取聊天ID和消息文本 message update.get(‘message’) if not message: continue chat_id message[‘chat’][‘id’] text message.get(‘text’) if not text: continue print(f“收到消息 [{chat_id}]: {text}”) # 分发处理 reply_text dispatcher.dispatch(text, chat_id) # 发送回复 if reply_text: client.send_message(chat_id, reply_text) print(f“发送回复: {reply_text[:50]}...”) # 避免过于频繁的请求短时间无消息时适当休眠 time.sleep(0.1) if __name__ “__main__”: main()至此一个不足200行核心逻辑部分的极简Telegram机器人框架就完成了。它包含了配置管理、API通信、技能系统和主循环。你可以运行python main.py然后用你的Bot Token去和机器人对话了。4. 功能扩展与实战踩坑从“能用”到“好用”上面的代码只是一个骨架真正让它变得“有用”还需要添加更多功能和处理边界情况。这里分享几个我实践中的扩展点和遇到的坑。4.1 实现代码执行技能这是一个高风险高价值的功能。安全是第一要务。我采用了Docker沙箱方案。创建code_runner_skill.pyimport docker import uuid from skill import Skill import re from config import config class CodeRunSkill(Skill): def __init__(self): self.client docker.from_env() self.image config.data[‘skills’][‘code_runner’][‘docker_image’] self.enabled config.data[‘skills’][‘code_runner’][‘enabled’] def can_handle(self, message_text): return self.enabled and re.match(r‘^/run\s(python|bash|javascript)\s’, message_text, re.IGNORECASE) def handle(self, message_text, chat_id, **kwargs): if not self.enabled: return “代码执行功能未启用。” # 解析语言和代码 match re.match(r‘^/run\s(\w)\s(.*)’, message_text, re.DOTALL) if not match: return “格式错误。用法: /run language code” lang, code match.groups() lang lang.lower() # 映射语言到Docker镜像内的执行命令 lang_cmd_map { ‘python’: [‘python’, ‘-c’], ‘bash’: [‘bash’, ‘-c’], ‘javascript’: [‘node’, ‘-e’], } if lang not in lang_cmd_map: return f“不支持的语言: {lang}。支持: {‘, ‘.join(lang_cmd_map.keys())}” cmd lang_cmd_map[lang] [code] container_name f“code_run_{uuid.uuid4().hex[:8]}” try: # 运行容器限制资源超时后自动停止 container self.client.containers.run( self.image, commandcmd, namecontainer_name, removeTrue, # 运行后自动删除容器 mem_limit‘50m’, # 内存限制 pids_limit50, # 进程数限制 network_mode‘none’, # 无网络访问 stdoutTrue, stderrTrue, detachFalse, # 同步执行获取输出 timeout10 # 执行超时时间 ) output container.decode(‘utf-8’) if isinstance(container, bytes) else container return f“执行结果:\n\n{output}\n” except docker.errors.ContainerError as e: return f“容器执行错误:\n\n{e.stderr.decode(‘utf-8’) if e.stderr else str(e)}\n” except docker.errors.ImageNotFound: return f“Docker镜像 {self.image} 不存在请先拉取。” except Exception as e: return f“执行过程中发生未知错误: {str(e)}”重要提示此代码执行方案仍有风险docker.from_env()需要主机有Docker守护进程且当前用户有权限。在生产环境中需要更严格的隔离、资源限制和审计日志。4.2 处理非文本消息与错误我们的主循环只处理了text消息。用户可能会发送图片、文档等。我们需要优雅地处理这些情况并加入更完善的错误处理。# 在主循环的更新处理部分进行增强 for update in updates: message update.get(‘message’) if not message: continue chat_id message[‘chat’][‘id’] # 处理文本消息 if ‘text’ in message: text message[‘text’] reply_text dispatcher.dispatch(text, chat_id) # 处理其他类型消息 elif ‘photo’ in message: reply_text “我收到了图片但我目前主要处理文本和代码哦。” elif ‘document’ in message: reply_text “我收到了文件但我目前主要处理文本和代码哦。” else: reply_text “抱歉我暂时无法处理这种类型的消息。” # 发送回复并加入重试机制 max_retries 3 for i in range(max_retries): try: client.send_message(chat_id, reply_text) break except Exception as send_e: if i max_retries - 1: print(f“向 {chat_id} 发送消息失败已达最大重试次数: {send_e}”) else: time.sleep(2 ** i) # 指数退避4.3 配置热重载与技能动态管理在机器人运行过程中我们可能想开启或关闭某个技能或者修改模型配置。我们可以实现一个简单的信号处理让机器人在收到SIGHUP信号时重新加载配置。import signal import threading from config import config class ReloadableConfig: def __init__(self, path): self.path path self.config None self.lock threading.Lock() self.load() def load(self): with self.lock: with open(self.path, ‘r’) as f: self.config yaml.safe_load(f) def get(self, key, defaultNone): with self.lock: # 支持点分键如 ‘skills.llm.enabled’ keys key.split(‘.’) value self.config for k in keys: if isinstance(value, dict): value value.get(k) else: return default return value if value is not None else default # 在主函数中 config_manager ReloadableConfig(‘config.yaml’) def reload_config(signum, frame): print(“收到重载配置信号...”) config_manager.load() # 这里可以通知各个技能重新初始化自己 # 例如: dispatcher.reload_skills() signal.signal(signal.SIGHUP, reload_config)这样修改config.yaml后只需要给机器人进程发送一个HUP信号kill -HUP pid配置就能生效无需重启。4.4 遇到的典型问题与解决问题一收不到消息或消息重复。排查检查Bot Token是否正确检查getUpdates的offset参数管理逻辑。如果offset没有正确更新下次轮询会拿到同样的消息。我的代码中self.last_update_id必须在成功处理一批消息后才更新为其中最大的update_id。解决确保在for update in updates:循环之后再更新last_update_id。问题二Ollama连接失败。排查首先确认Ollama服务是否运行curl http://localhost:11434。其次检查配置中的base_url和model名称是否正确。模型名称需要是已通过ollama pull拉取到本地的。解决在LLMSkill的handle方法中加入更详细的错误捕获和提示如区分连接错误和模型不存在错误。问题三Docker代码执行权限问题。排查运行docker ps看命令是否正常。普通用户可能不在docker用户组。解决将当前用户加入docker组sudo usermod -aG docker $USER然后需要重新登录生效。或者考虑使用基于API的代码执行服务如Piston作为更轻量、更安全的替代方案。问题四机器人响应慢。排查getUpdates的timeout参数设置过长会阻塞主循环某个技能处理耗时过长如LLM生成。解决将timeout设为10-20秒的合理值。对于耗时技能可以考虑将其放入线程池异步执行先回复一个“正在处理”的提示处理完成后再发送最终结果。但这会稍微增加代码复杂度违背“极简”初衷可根据需要权衡。5. 部署与进阶思考容器化与技能市场让这个机器人7x24小时运行并方便地扩展就需要考虑部署。5.1 使用Docker容器化部署创建一个Dockerfile将我们的机器人打包成镜像便于在任何支持Docker的环境运行。FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [“python”, “main.py”]requirements.txt内容requests2.28.0 pyyaml6.0 docker6.0.0构建并运行docker build -t mini-moltbot . docker run -d \ --name my-moltbot \ -v /var/run/docker.sock:/var/run/docker.sock \ # 挂载docker socket以便在容器内运行代码容器 -v $(pwd)/config.yaml:/app/config.yaml \ -e BOT_TOKENyour_token_here \ mini-moltbot警告挂载Docker socket (-v /var/run/docker.sock:/var/run/docker.sock) 会将主机Docker守护进程的控制权暴露给容器存在安全风险。仅建议在可信的测试环境中使用。生产环境应考虑更安全的隔离方案如使用独立的Docker API over TCP并配置认证。5.2 技能市场的构想我们目前的技能是硬编码在main.py中注册的。一个更优雅的设计是“技能发现”机制。我们可以约定一个技能目录如skills/每个技能是一个独立的Python文件或包。主程序启动时扫描这个目录动态导入并注册所有继承自Skill基类的类。import importlib.util import os def load_skills_from_folder(folder_path): skills [] for filename in os.listdir(folder_path): if filename.endswith(‘.py’) and not filename.startswith(‘__’): module_name filename[:-3] spec importlib.util.spec_from_file_location(module_name, os.path.join(folder_path, filename)) module importlib.util.module_from_spec(spec) spec.loader.exec_module(module) for attr_name in dir(module): attr getattr(module, attr_name) if isinstance(attr, type) and issubclass(attr, Skill) and attr ! Skill: skills.append(attr()) return skills # 在主程序中 skill_folder “skills” if os.path.exists(skill_folder): auto_skills load_skills_from_folder(skill_folder) for skill in auto_skills: dispatcher.register_skill(skill)这样要添加一个新技能只需要在skills/文件夹里丢一个符合规范的Python文件即可实现了真正的插件化。5.3 从“tele-code”到更多可能这个400行代码的框架其价值在于清晰地勾勒出了一个可扩展的聊天机器人内核。它被称作“tele-code”大概是因为它最初聚焦于处理代码相关任务。但基于这个框架你可以轻松地将其改造成任何你想要的自动化助手智能家居控制添加一个技能解析“打开客厅灯”这样的自然语言调用Home Assistant或米家API。个人知识库问答添加一个技能将用户问题向量化从本地ChromaDB或Milvus中检索相关文档片段交给LLM生成答案。服务器监控添加一个技能响应/status命令执行df -h或docker ps等命令并返回结果需妥善处理权限和安全。它的边界只取决于你注册了什么样的技能。这个过程中最重要的收获不是这400行代码本身而是对“消息流-技能匹配-执行-反馈”这一机器人核心范式的透彻理解。当你需要更复杂的功能时你知道该在哪个环节进行增强而不是面对一个庞大的黑盒框架无从下手。