ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀,每一条都经过真实交付验证。

AstrBot插件开发:从零构建可热加载的天气查询模块

2026/9/19 18:49:36 拓冰建站 浏览量
AstrBot插件开发:从零构建可热加载的天气查询模块 1. AstrBot不是另一个聊天机器人它是你手里的“插件操作系统”AstrBot这个词最近在开源机器人圈子里冒得很快但很多人第一眼看到下意识就把它归类成“又一个QQ/微信机器人框架”——这恰恰是踩进的第一个认知坑。它和NoneBot、CoolQ、Mirai这些传统Bot框架有本质区别AstrBot不处理消息收发、不管理连接池、不封装协议层它只做一件事把插件变成可热加载、可独立生命周期管理、可跨平台运行的Python模块单元。你可以把它理解成一个轻量级的“插件容器OS”而你的天气查询插件就是跑在这个OS上的第一个原生应用。我第一次接触AstrBot是在给某高校社团做自动化值班提醒系统时。当时用的是NoneBot2写完天气插件后发现一个问题每次改一行代码就得重启整个Bot服务所有在线用户都会收到“机器人已离线”的提示更麻烦的是天气API密钥、城市缓存、错误重试策略这些配置全混在main.py里和签到、查课表、查空教室这些功能搅在一起动一处怕崩一片。直到我把核心逻辑抽出来用AstrBot的PluginManager重新组织才真正体会到什么叫“插件即服务”。它的设计哲学很朴素每个插件必须自包含、自启动、自销毁。这意味着你的天气插件不需要知道Bot用的是WebSocket还是HTTP轮询不需要关心消息是来自QQ群还是Telegram频道甚至不需要知道当前Bot是否在线——它只管三件事监听特定指令比如“今天北京天气”、调用天气API、返回结构化响应。其余所有胶水逻辑由AstrBot Runtime统一兜底。这也是为什么标题强调“从零编写”——它不是教你如何在现有Bot里加个函数而是带你亲手造一个能被AstrBot识别、加载、调度、卸载的完整插件包。这个过程会强制你思考插件的入口点在哪不是main()而是Plugin类的on_load()配置文件该放哪不是config.yaml硬编码而是通过AstrBot的ConfigManager注入错误怎么上报不是print()而是调用self.logger.error()如何避免阻塞主线程不是time.sleep()而是用asyncio.to_thread()或aiohttp这些细节恰恰是大多数教程跳过的“脏活”。但正是这些脏活决定了你的插件是能稳定跑一周还是上线两小时就被用户投诉“机器人卡死了”。提示AstrBot官方文档里有一句容易被忽略的话“插件不应持有全局状态”。这句话背后藏着一个血泪教训——我曾见过一个插件用全局字典缓存城市ID结果在多进程部署时每个Worker进程都维护一份副本导致缓存不同步用户查上海天气却返回了深圳数据。后面我会专门拆解这个陷阱。现在我们手里没有现成的模板没有一键生成脚手架只有Python解释器和AstrBot源码。接下来要做的是像搭积木一样一块一块拼出这个天气插件的骨架、血肉和神经。2. 插件结构不是目录树而是运行时契约很多开发者一上来就猛敲mkdir -p astrbot_weather/{init.py,plugin.py,config.yaml}以为目录建对了插件就成功了一半。错。AstrBot识别插件靠的不是文件夹名字而是插件模块内部是否满足一套严格的运行时契约。这套契约由四个核心接口构成缺一不可且顺序不能乱。下面我用实际代码逐行拆解告诉你为什么少一个方法插件就永远进不了AstrBot的插件列表。2.1 插件元信息不是装饰器是注册凭证# plugin.py from astrbot.core import Plugin from astrbot.core.model.provider import Provider from astrbot.core.utils import logger class WeatherPlugin(Plugin): def __init__(self, provider: Provider, **kwargs): super().__init__(provider, **kwargs) self.logger logger这段代码里WeatherPlugin继承自astrbot.core.Plugin这是契约的第一块基石。但注意__init__方法签名必须带provider: Provider参数且必须调用super().__init__()。这不是形式主义——AstrBot在加载插件时会反射调用Plugin.__init__并传入当前Bot的Provider实例负责消息分发、事件总线等。如果你删掉provider参数或者没调用父类初始化AstrBot会在日志里默默记录[PluginLoader] Failed to instantiate plugin WeatherPlugin然后跳过这个插件连错误都不抛给你。我踩过这个坑。当时为了快速测试我把provider参数改成*args, **kwargs心想“反正我暂时不用”。结果插件一直显示“未启用”翻遍日志也找不到原因。最后用pdb断点跟踪才发现AstrBot的PluginLoader在实例化时硬编码了provider参数名不匹配就直接跳过。这种设计看似死板实则是为了保证所有插件都能访问统一的消息分发通道避免各自为政。2.2 生命周期钩子on_load()不是启动函数是资源预热场def on_load(self): self.logger.info(天气插件开始加载...) # 1. 加载配置 self.config self.get_config() # 2. 初始化API客户端 self.api_client self._init_weather_api() # 3. 注册指令监听 self.register_command(天气, self.handle_weather_query) self.register_command(weather, self.handle_weather_query) self.logger.info(天气插件加载完成)on_load()是契约的第二块基石也是最容易被误解的地方。很多人以为这里就是写业务逻辑的起点于是把API调用、数据库连接全堆进去。大错特错。on_load()的唯一使命是为插件准备好运行所需的最小资源集合并向AstrBot注册自己的能力声明。上面代码里三件事每一件都有严格边界self.get_config()调用AstrBot内置的配置管理器读取插件专属配置稍后详解。不能自己open()读文件否则配置热更新失效。self._init_weather_api()初始化API客户端但绝不发起真实请求。这里只是创建aiohttp.ClientSession实例或设置requests.Session的默认headers。真实调用留到handle_weather_query()里。self.register_command()向AstrBot的命令路由表注册指令映射。注意这里注册的是字符串指令天气和回调函数self.handle_weather_query的绑定关系不是立即执行。AstrBot后续收到消息时会根据这个表决定哪个插件来处理。注意on_load()必须是同步函数不能用async def。因为AstrBot的插件加载器是同步执行的如果这里写await会导致整个Bot启动失败。我曾用asyncio.sleep(0.1)模拟网络延迟结果Bot卡在启动阶段日志里只有一行INFO: Application startup complete.再无下文。排查了三小时才发现是这个钩子函数类型错了。2.3 指令处理器handle_weather_query不是普通函数是异步事件处理器async def handle_weather_query(self, message: str, context: dict): 处理天气查询指令 message: 用户发送的原始消息如天气 北京 context: 上下文字典含sender_id, platform等元信息 # 1. 解析城市名 city self._extract_city_from_message(message) if not city: return 请告诉我你想查询的城市例如天气 上海 # 2. 调用天气API异步 try: weather_data await self._fetch_weather_data(city) except Exception as e: self.logger.error(f获取{city}天气数据失败: {e}) return f抱歉查询{city}天气时遇到问题请稍后再试 # 3. 格式化响应 return self._format_weather_response(weather_data, city)这是契约的第三块基石也是业务逻辑的核心载体。关键点在于它必须是async def定义的协程函数且参数签名固定为(self, message: str, context: dict)。AstrBot的消息分发器在匹配到指令后会以await plugin.handle_weather_query(message, context)方式调用它。这里藏着两个实战技巧城市名提取必须健壮用户可能说“北京天气怎么样”、“查一下深圳的天气”、“上海明天热不热”。我最初用message.split()[1]结果用户发“天气”两个字就报错。后来改用正则re.search(r(?:天气|weather)\s*(\S), message, re.I)覆盖了95%的口语变体。API调用必须带超时和重试免费天气API经常抖动。我在_fetch_weather_data()里用了asyncio.wait_for()包裹超时设为8秒并配合指数退避重试最多2次。实测下来比单纯try-except稳定得多。2.4 卸载钩子on_unload()不是清理函数是优雅退出协议def on_unload(self): self.logger.info(天气插件开始卸载...) # 关闭API客户端 if hasattr(self, api_client) and self.api_client: try: # 如果是aiohttp.ClientSession需显式关闭 if hasattr(self.api_client, close): asyncio.create_task(self.api_client.close()) except Exception as e: self.logger.warning(f关闭API客户端时警告: {e}) self.logger.info(天气插件卸载完成)这是契约的最后一块基石。on_unload()在插件被禁用或Bot重启时触发。它的作用不是“清理垃圾”而是履行与AstrBot Runtime的退出协议释放独占资源如网络连接、文件句柄通知上游服务如注销Webhook并确保自身状态可被安全丢弃。重点来了这里不能用await因为on_unload()是同步函数但aiohttp.ClientSession的close()是异步的。我的解决方案是asyncio.create_task()——把它扔进事件循环让Runtime自己去处理。如果直接self.api_client.close()会报RuntimeError: Task got bad yield: coroutine object ClientSession.close at 0x...。这个细节官方文档没写但源码里PluginManager.unload_plugin()方法明确要求on_unload()返回None。现在这个插件的骨架已经立住了。它不是一个Python脚本而是一个符合AstrBot运行时契约的、可被动态管理的软件组件。下一步我们要给它装上眼睛配置、心脏API客户端和嘴巴响应格式化。3. 配置不是ini文件而是插件的“环境变量沙箱”AstrBot的配置体系是它和其它Bot框架拉开差距的关键设计。它不让你把API密钥写死在代码里也不鼓励你用os.getenv()读取全局环境变量而是提供了一套插件专属的配置沙箱机制。这个沙箱有三个核心特性层级隔离、热更新支持、类型安全校验。下面我用真实配置文件和代码展示如何让天气插件既安全又灵活。3.1 配置文件结构config.yaml不是随便写的文本是Schema契约# config.yaml # 插件配置根节点必须与插件名一致此处为weather weather: # API服务提供商支持和风、心知、OpenWeatherMap provider: qweather # API密钥敏感信息建议用环境变量注入 api_key: ${WEATHER_API_KEY} # 默认城市当用户未指定时使用 default_city: 北京 # 缓存有效期秒减少重复API调用 cache_ttl: 1800 # 城市名称映射表解决用户说“魔都”“羊城”等别名 city_aliases: 魔都: 上海 羊城: 广州 春城: 昆明 冰城: 哈尔滨 # 响应风格简洁/详细/emoji response_style: emoji这个文件看起来像普通YAML但每一行都承载着契约根节点名weather必须与插件名完全一致。AstrBot在加载插件时会根据插件模块名这里是astrbot_weather自动匹配config.yaml中同名的配置段。如果写成weather_plugin配置就永远无法注入。${WEATHER_API_KEY}不是字符串而是环境变量占位符。AstrBot的ConfigManager在解析时会自动替换为系统环境变量值。这样你的代码里永远看不到明文密钥部署时只需export WEATHER_API_KEYxxx即可。cache_ttl: 1800不是数字而是带单位的配置项。AstrBot会校验其类型是否为int如果不是比如写成1800字符串会在日志里警告[Config] Invalid type for cache_ttl: expected int, got str但不会崩溃——这是它容错设计的体现。我最初把api_key直接写成明文结果Git提交后被同事发现紧急撤回。后来改用环境变量但忘了在Docker Compose里加environment:字段导致线上插件一直报“API密钥为空”。这个教训让我明白配置沙箱的价值不在于写起来多方便而在于它强制你把“密钥管理”这件事从代码层提升到部署层。3.2 配置加载get_config()不是读文件是运行时注入def on_load(self): # ... 其他代码 self.config self.get_config() # 验证必要配置项 if not self.config.get(api_key): self.logger.error(天气插件配置缺失api_key 不能为空) raise ValueError(API密钥未配置请检查config.yaml或环境变量WEATHER_API_KEY) if not self.config.get(default_city): self.logger.warning(天气插件未配置default_city将使用北京作为默认城市) self.config[default_city] 北京self.get_config()是AstrBot提供的魔法方法。它不是简单地yaml.safe_load(open(config.yaml))而是自动合并多层级配置如base.yamlprod.yaml自动解析环境变量占位符${VAR}自动进行基础类型转换YAML里的1800转为Python int自动注入插件专属命名空间只返回weather:下的子树关键点在于get_config()返回的是一个深拷贝字典不是引用。这意味着你在插件里修改self.config[cache_ttl] 3600不会影响其他插件也不会污染全局配置。这是沙箱隔离性的技术保障。提示get_config()在on_load()里调用一次就够了。不要在每次handle_weather_query()里都调用它——那会反复解析YAML浪费CPU。我曾做过性能测试在高并发场景下频繁调用get_config()会使QPS下降12%。正确做法是加载时存为实例变量后续直接读取。3.3 配置热更新reload_config()不是重读文件是运行时热切换def on_reload(self): 配置热更新钩子AstrBot 0.8.0 支持 new_config self.get_config() # 检查关键配置是否变更 if new_config.get(api_key) ! self.config.get(api_key): self.logger.info(检测到API密钥变更正在重置API客户端...) self._reset_api_client(new_config[api_key]) if new_config.get(cache_ttl) ! self.config.get(cache_ttl): self.logger.info(f缓存有效期更新为 {new_config[cache_ttl]} 秒) self.cache_ttl new_config[cache_ttl] self.config new_config self.logger.info(配置热更新完成)AstrBot 0.8.0版本引入了on_reload()钩子这是配置沙箱的终极形态。当config.yaml被修改并保存后AstrBot会自动触发此方法无需重启Bot。但注意它不是自动生效的你需要自己实现配置变更的响应逻辑。上面代码展示了两个典型场景API密钥变更需要重建API客户端因为旧密钥已失效。缓存时间变更只需更新实例变量下次查询自然生效。我实测过从修改配置文件到插件响应新配置平均耗时230ms基于inotify监听。比重启Bot平均4.2秒快18倍。这对于需要频繁调整参数的运维场景价值巨大。现在插件有了可配置的“眼睛”能看清用户意图也能适应不同环境。下一步我们要给它装上“心脏”——一个稳定、高效、带熔断的天气API客户端。4. 天气API客户端不是requests.get()而是带熔断的异步服务网关市面上的天气插件教程十有八九用requests.get()配个URL就完事。这在本地测试时没问题但放到生产环境会立刻暴露三个致命缺陷同步阻塞一个API请求卡住整个Bot消息队列就堵死无熔断API服务商宕机时插件持续重试拖垮Bot内存无缓存同一城市被查100次就发100次HTTP请求白费流量还易被限流。AstrBot的天气插件必须超越这个层次。我们要构建一个异步、带熔断、带LRU缓存、带错误分类的日志化API客户端。下面用真实代码一步步组装这个“心脏”。4.1 客户端初始化_init_weather_api()不是创建session是构建服务网关def _init_weather_api(self): 初始化天气API客户端 # 1. 创建aiohttp ClientSession复用TCP连接 connector aiohttp.TCPConnector( limit10, # 最大并发连接数 limit_per_host5, # 单域名最大连接数 keepalive_timeout30, # 连接保活时间 ) timeout aiohttp.ClientTimeout( total10, # 总超时含DNS、连接、读取 connect5, # 连接超时 sock_read8, # Socket读取超时 ) self.session aiohttp.ClientSession( connectorconnector, timeouttimeout, headers{ User-Agent: AstrBot-Weather/1.0, Accept: application/json } ) # 2. 初始化熔断器基于tenacity库 from tenacity import RetryError, retry, stop_after_attempt, wait_exponential self.retry_strategy retry( stopstop_after_attempt(3), # 最多重试3次 waitwait_exponential(multiplier1, min1, max10), # 指数退避 reraiseTrue # 抛出最后一次异常 ) # 3. 初始化LRU缓存内存缓存非Redis from functools import lru_cache self._fetch_weather_data_cached lru_cache(maxsize100)( self._fetch_weather_data_uncached ) return self.session这段初始化代码每一行都在解决一个生产痛点TCPConnector的limit10防止插件耗尽Bot的全部网络连接ClientTimeout的total10确保单次请求不会无限等待tenacity.retry的stop_after_attempt(3)避免API永久故障时无限重试lru_cache(maxsize100)让热门城市如北京、上海的查询直接走内存毫秒级返回。注意lru_cache装饰的是_fetch_weather_data_uncached而不是_fetch_weather_data。因为后者是async def不能直接被lru_cache装饰。我们用一个同步包装函数来桥接具体实现见下一节。4.2 核心API调用_fetch_weather_data()不是发请求是服务编排流水线async def _fetch_weather_data(self, city: str) - dict: 获取城市天气数据带缓存、熔断、错误分类 返回标准化的天气字典结构统一便于后续格式化 # 1. 尝试从缓存读取同步操作极快 cache_key f{city}_{self.config.get(provider, qweather)} try: cached self._fetch_weather_data_cached(cache_key) if cached: self.logger.debug(f缓存命中: {city}) return cached except Exception as e: self.logger.debug(f缓存读取异常: {e}) # 2. 执行带熔断的API调用 try: result await self.retry_strategy(self._fetch_weather_data_uncached)(cache_key) # 3. 写入缓存异步避免阻塞 asyncio.create_task(self._cache_weather_data(cache_key, result)) return result except RetryError as e: self.logger.error(fAPI重试3次后仍失败: {city}, 原因: {e}) raise except aiohttp.ClientError as e: self.logger.error(f网络层错误: {city}, {e}) raise except asyncio.TimeoutError: self.logger.error(fAPI请求超时: {city}) raise except Exception as e: self.logger.error(f未知错误: {city}, {e}) raise async def _fetch_weather_data_uncached(self, cache_key: str) - dict: 无缓存的原始API调用被lru_cache和retry装饰 city cache_key.split(_)[0] provider self.config.get(provider, qweather) if provider qweather: return await self._fetch_qweather_data(city) elif provider openweathermap: return await self._fetch_openweathermap_data(city) else: raise ValueError(f不支持的天气服务商: {provider}) async def _cache_weather_data(self, cache_key: str, data: dict): 异步写入缓存避免阻塞主流程 # 这里可以扩展为写入Redis当前用内存缓存 pass这个方法是整个客户端的中枢。它把一次天气查询拆解成清晰的流水线缓存探查先查LRU缓存命中则秒回不走网络熔断执行调用被tenacity.retry装饰的_fetch_weather_data_uncached自动重试缓存写入成功后用asyncio.create_task()异步写入不影响主流程响应速度错误分类不同异常类型网络错误、超时、业务错误打不同日志级别便于监控。我特意把_fetch_weather_data_uncached单独拆出来是因为它要同时被lru_cache同步和tenacity.retry异步装饰。Python的装饰器链要求同步装饰器必须在异步装饰器外层否则会报TypeError: object NoneType cant be used in await expression。这个细节不实际写过的人很难注意到。4.3 服务商适配_fetch_qweather_data()不是拼URL是协议适配器async def _fetch_qweather_data(self, city: str) - dict: 和风天气API适配器v7版本 # 1. 获取城市ID需先调用地理编码API location_url https://geoapi.qweather.com/v2/city/lookup params { key: self.config[api_key], location: city } async with self.session.get(location_url, paramsparams) as resp: if resp.status ! 200: raise Exception(f地理编码API返回{resp.status}: {await resp.text()}) loc_data await resp.json() if not loc_data.get(location): raise ValueError(f未找到城市: {city}) city_id loc_data[location][0][id] # 2. 获取实时天气 weather_url https://devapi.qweather.com/v7/weather/now params { key: self.config[api_key], location: city_id } async with self.session.get(weather_url, paramsparams) as resp: if resp.status ! 200: raise Exception(f天气API返回{resp.status}: {await resp.text()}) weather_data await resp.json() # 3. 标准化返回结构统一字段名便于后续格式化 return { city: city, temperature: float(weather_data[now][temp]), condition: weather_data[now][textDay], humidity: int(weather_data[now][humidity]), wind_direction: weather_data[now][windDir], wind_speed: float(weather_data[now][windScale]), update_time: weather_data[lastUpdate] }这里展示了真正的“协议适配器”思维。和风天气API需要两步先查城市ID再查天气。而OpenWeatherMap一步就能搞定。如果把所有逻辑写在_fetch_weather_data_uncached里代码会变得臃肿难维护。所以我们为每个服务商单独写适配器它们只负责构造正确的URL和参数处理服务商特有的错误码如和风的403表示密钥无效将原始JSON映射为统一的标准化字典。这个标准化字典就是插件的“内部协议”。无论后端换哪家API_format_weather_response()方法都不用改——它只认temperature、condition这些字段。这就是架构解耦的力量。现在“心脏”已经装好能稳定跳动、自我保护、智能缓存。最后一步我们要给插件装上“嘴巴”让它能把冷冰冰的数据变成用户爱看的、带温度的响应。5. 响应格式化不是print()而是多模态消息渲染引擎用户输入“天气 北京”插件返回{temperature: 25.3, condition: 晴, ...}这显然不行。AstrBot的handle_weather_query()方法最终必须返回一个字符串这个字符串会被Bot框架原样发送给用户。但“原样发送”不等于“随便拼接”。一个专业的天气插件应该根据配置、用户习惯、平台特性动态生成最合适的响应。这就需要一套多模态消息渲染引擎。5.1 响应策略_format_weather_response()不是字符串拼接是策略模式调度器def _format_weather_response(self, weather_data: dict, city: str) - str: 根据配置的response_style生成不同风格的响应 style self.config.get(response_style, simple) if style emoji: return self._format_emoji_style(weather_data, city) elif style detailed: return self._format_detailed_style(weather_data, city) else: # default simple return self._format_simple_style(weather_data, city) def _format_simple_style(self, data: dict, city: str) - str: 简洁风格城市 温度 天气状况 return f{city} {data[temperature]}°C {data[condition]} def _format_emoji_style(self, data: dict, city: str) - str: Emoji风格用图标增强可读性 # 温度emoji映射 temp_emoji if data[temperature] 30 else \ ☀️ if data[temperature] 15 else \ ☁️ if data[temperature] 5 else ❄️ # 天气状况emoji映射 cond_emoji { 晴: ☀️, 多云: ☁️, 阴: ⛅, 小雨: ️, 中雨: ️, 大雨: ⛈️, 雷阵雨: ⚡, 雪: ❄️ }.get(data[condition], ️) return f{city} {temp_emoji}{data[temperature]}°C {cond_emoji}{data[condition]} def _format_detailed_style(self, data: dict, city: str) - str: 详细风格包含湿度、风向、风速、更新时间 update_time datetime.fromisoformat(data[update_time].replace(Z, 00:00)) return (f【{city}天气】\n f️ 温度: {data[temperature]}°C\n f☁️ 天气: {data[condition]}\n f 湿度: {data[humidity]}%\n f 风向: {data[wind_direction]}\n f 风力: {data[wind_speed]}级\n f⏰ 更新: {update_time.strftime(%H:%M)})这个设计体现了“开闭原则”新增一种响应风格比如Markdown表格、图片卡片只需添加一个_format_xxx_style()方法并在_format_weather_response()里加一个分支完全不用动原有逻辑。我上线后有用户反馈“Emoji太多看着累”我们立刻加了个text风格纯文字无图标当天就推给了他。提示datetime.fromisoformat()这里有个坑。和风API返回的时间是ISO 8601格式但末尾带ZUTC时区而Python 3.6的fromisoformat()不支持Z必须替换成00:00。这个细节不查文档根本不知道我调试时卡在ValueError: Invalid isoformat string上半小时。5.2 平台适配send_message()不是send()是渠道感知分发器虽然handle_weather_query()返回字符串但AstrBot的Runtime会根据消息来源平台QQ、Telegram、Discord自动做适配。不过有些平台特性需要插件主动感知async def handle_weather_query(self, message: str, context: dict): # ... 解析、调用API ... response self._format_weather_response(weather_data, city) # 平台特化处理 platform context.get(platform, unknown) if platform qq: # QQ平台支持CQ码可加粗关键信息 response response.replace(【, [CQ:at,qq]).replace(】, ) elif platform telegram: # Telegram支持Markdown可加粗 response response.replace(【, **).replace(】, **) elif platform discord: # Discord支持Embed但插件层不直接构造交由Runtime处理 pass return responseAstrBot的Runtime会接管最终的消息发送但插件可以通过context字典拿到platform标识做轻量级适配。比如在QQ里用CQ码用户在Telegram里用Markdown加粗让响应更原生。注意这里不做重构造如Discord Embed因为那是Runtime的职责插件只负责提供语义清晰的文本。5.3 错误响应不是裸露异常是用户友好的降级策略async def handle_weather_query(self, message: str, context: dict): try: # ... 正常流程 ... except ValueError as e: # 用户输入错误如城市不存在 return f❌ {str(e)}请确认城市名称是否正确 except Exception as e: # 系统错误API故障、网络超时 self.logger.exception(天气查询内部错误) # 降级返回缓存数据如果有 cache_key f{city}_{self.config.get(provider, qweather)} try: cached self._fetch_weather_data_cached(cache_key) if cached: self.logger.info(f降级使用缓存数据: {city}) return self._format_weather_response(cached, city) 数据可能已过期 except: pass return ⚠️ 服务器繁忙请稍后再试~这才是专业插件的错误处理。它分三层用户错误清晰告知问题所在“城市不存在”引导用户修正系统错误记录完整traceback便于排查降级策略尝试用缓存数据兜底哪怕过期也比直接报错强。我上线第一天和风API就挂了2小时。因为有这个降级用户只看到“数据可能已过期”的提示没人投诉。而隔壁用requests直连的插件全在刷“机器人坏了”。现在从插件结构、配置沙箱、API客户端到响应引擎整个天气插件已经完整闭环。它不再是一个脚本而是一个具备生产级鲁棒性的软件模块。最后我想分享一个在真实项目中验证过的部署 checklist。6. 生产部署 checklist不是pip install而是可审计的交付包写完插件本地测试通过很多人就直接pip install -e .完事。但在团队协作或生产环境这远远不够。一个可交付的AstrBot插件必须满足五个可审计条件可复现构建、可验证签名、可追溯版本、可灰度发布、可一键回滚。下面是我给客户交付时必做的七项检查。6.1 构建可复现requirements.txt 不是 pip freeze是精简依赖锁