1. 项目概述从零构建一个AI驱动的自动化技能库最近在折腾一个叫openclaw-skill-imsg的项目这名字听起来有点技术范儿但说白了它的核心目标就是打造一个能处理即时消息IM的AI智能体技能库。想象一下你有一个AI助手它不仅能和你聊天还能帮你自动回复微信消息、整理邮件里的关键信息、或者在钉钉群里自动收集日报。openclaw-skill-imsg就是为这类场景提供“技能”的工具箱。它不是一个完整的聊天机器人而是一个专注于“消息处理”这个垂直领域的技能集合你可以把它理解为一个乐高积木包开发者可以从中挑选合适的模块快速组装出能理解、分析和响应各种即时消息的AI应用。这个项目特别适合两类人一是对AI应用开发感兴趣的工程师尤其是想将大语言模型LLM能力落地到具体业务场景如客服、办公自动化的实践者二是那些被重复性消息处理工作困扰的团队希望通过自动化提升效率。我自己在尝试将一些日常工作流程自动化时发现市面上要么是过于庞大、学习成本高的框架要么是功能单一、难以扩展的小工具。openclaw-skill-imsg的设计理念恰好踩在了这个平衡点上——它提供了一套标准化的技能开发接口让开发者可以专注于业务逻辑本身而不用重复造轮子去处理消息的接收、解析、上下文管理等底层杂务。2. 核心架构与设计思路拆解2.1 为什么是“技能库”而非“完整机器人”在开始深入代码之前理解这个设计选择至关重要。很多初学者会困惑为什么不直接做一个完整的机器人答案在于“解耦”和“复用”。一个完整的聊天机器人系统通常包含多个紧密耦合的组件消息接入层、自然语言理解NLU引擎、对话管理DM模块、技能执行器以及回复生成层。这种架构虽然完整但非常笨重。如果你想为机器人增加一个“查询天气”的技能你可能需要修改NLU的意图识别、DM的状态机甚至改动回复模板。openclaw-skill-imsg采用了截然不同的思路。它假设你已经有了一个可以接收和发送消息的“运行时环境”比如一个微信机器人框架、一个钉钉Webhook服务器这个环境负责最底层的协议通信。而openclaw-skill-imsg则专注于提供运行在这个环境之上的“技能”。每个技能都是一个独立的、功能完备的单元它接收结构化的消息请求执行逻辑可能调用LLM然后返回结构化的响应。这种设计带来了几个显著优势低侵入性你不需要推翻现有的消息处理框架只需要将其作为技能加载进去即可。高可插拔性技能的开发、测试、部署可以独立进行。今天加一个“翻译”技能明天移除一个“讲笑话”技能整个系统无需重启或大规模重构。技术栈无关性技能内部可以用Python、Node.js甚至Java编写只要它遵循统一的输入输出接口。消息运行时环境也可以用任何语言实现。便于共享与生态建设一个设计良好的“查询会议室”技能可以被公司内不同的聊天机器人项目复用甚至开源给社区。2.2 核心组件交互模型基于上述思路openclaw-skill-imsg的核心架构通常包含以下几个关键部分我们可以通过一个简单的时序图来理解注意以下为逻辑描述非实际代码技能Skill这是核心单元。每个技能都有唯一的标识符如weather_query并实现一个标准的execute方法。这个方法接收一个Context对象包含用户消息、会话历史、用户身份等信息执行逻辑后返回一个SkillResponse对象。技能管理器Skill Manager负责技能的注册、发现、加载和生命周期管理。当运行时环境收到一条消息后它会询问技能管理器“当前有哪些技能可以处理这条消息”技能管理器会根据技能的元信息如触发关键词、意图匹配规则进行筛选。运行时环境Runtime这是项目的“外壳”它负责与具体的IM平台如微信、飞书、Slack对接处理网络协议、认证、消息编解码等。它调用技能管理器来获取合适的技能并执行。上下文Context这是一个贯穿始终的数据载体。它不仅仅包含当前消息的文本还可能包含丰富的元数据发送者信息、群组信息、消息类型文本、图片、文件、甚至是经过上游NLU处理后的结构化结果如意图、实体。技能可以读取和修改上下文从而实现多轮对话和状态保持。一个典型的工作流程如下用户发送消息“今天北京天气怎么样” - 运行时环境接收消息封装成Context- 运行时环境调用技能管理器的匹配方法 - 技能管理器遍历所有已注册技能发现weather_query技能注册的触发关键词包含“天气”于是返回该技能 - 运行时环境调用weather_query.execute(context)- 技能内部解析context中的城市实体“北京”调用外部天气API - 技能将结果格式化填入SkillResponse并返回 - 运行时环境将SkillResponse中的内容转换为IM平台所需的格式如XML、JSON并发送给用户。注意这里有一个关键设计决策——技能匹配的粒度。是让每个技能自己声明一个匹配函数更灵活还是由技能管理器基于统一的规则如关键词、正则表达式来匹配更一致openclaw-skill-imsg通常倾向于后者即技能在注册时提供静态的匹配规则由管理器统一处理。这保证了匹配逻辑的可预测性和可管理性避免技能代码过于臃肿。3. 技能开发实战从零编写一个“待办事项”技能理论讲得再多不如动手写一个。我们以开发一个简单的“待办事项管理”技能为例看看如何基于openclaw-skill-imsg的模式进行开发。这个技能允许用户通过自然语言添加、查看和删除待办事项。3.1 定义技能接口与数据模型首先我们需要定义技能必须遵守的“契约”。虽然openclaw-skill-imsg项目本身可能提供了基础接口但理解其设计至关重要。一个典型的技能基类可能长这样# skill_base.py from abc import ABC, abstractmethod from typing import Dict, Any, Optional from dataclasses import dataclass dataclass class SkillContext: 技能执行的上下文包含所有必要信息 user_id: str # 用户唯一标识 message_text: str # 原始消息文本 message_type: str # 消息类型如 “text”, “image” session_id: str # 会话ID用于区分私聊和群聊 entities: Dict[str, Any] None # 从消息中提取的实体如时间、地点 intent: str None # 消息的意图如 “add_todo” # ... 其他平台特定字段 dataclass class SkillResponse: 技能执行后的响应 success: bool # 执行是否成功 message: str # 返回给用户的文本消息 data: Optional[Dict[str, Any]] None # 附加数据可用于后续处理 need_further_input: bool False # 是否需要用户进一步输入多轮对话 class BaseSkill(ABC): 所有技能的基类 property abstractmethod def name(self) - str: 技能的唯一名称如 ‘todo_manager’ pass property abstractmethod def description(self) - str: 技能的简短描述 pass property def trigger_keywords(self) - list: 触发该技能的关键词列表用于快速匹配 return [] # 默认空可由子类覆盖 abstractmethod async def execute(self, context: SkillContext) - SkillResponse: 执行技能的核心逻辑 pass def can_handle(self, context: SkillContext) - bool: 判断该技能是否能处理当前上下文。 默认实现检查关键词子类可覆盖以实现更复杂的逻辑如意图匹配。 if self.trigger_keywords: return any(keyword in context.message_text for keyword in self.trigger_keywords) return False这个基类定义了技能的基本骨架一个名字、一段描述、一些触发关键词以及最核心的execute异步方法。SkillContext和SkillResponse是两个重要的数据容器它们标准化了技能与运行时环境之间的数据交换格式。3.2 实现待办事项管理技能接下来我们实现具体的TodoSkill。为了简化我们使用一个内存中的字典来存储待办事项。在实际生产中你应该连接数据库。# todo_skill.py import re from typing import List from skill_base import BaseSkill, SkillContext, SkillResponse class TodoSkill(BaseSkill): def __init__(self): # 用字典模拟存储 {user_id: [todo_item1, todo_item2, ...]} self._storage {} # 定义技能支持的指令模式 self._add_pattern re.compile(r(添加|增加|新建)待办[:]?\s*(.)) self._list_pattern re.compile(r(查看|列出|我的)待办) self._delete_pattern re.compile(r(删除|完成|移除)待办[:]?\s*(\d)) # 按索引删除 property def name(self) - str: return todo_manager property def description(self) - str: return 管理个人待办事项支持添加、查看和删除。 property def trigger_keywords(self) - list: return [待办, todo] async def execute(self, context: SkillContext) - SkillResponse: user_id context.user_id message context.message_text.strip() # 1. 尝试匹配“添加待办” match_add self._add_pattern.search(message) if match_add: todo_content match_add.group(2) return await self._add_todo(user_id, todo_content) # 2. 尝试匹配“查看待办” match_list self._list_pattern.search(message) if match_list: return await self._list_todos(user_id) # 3. 尝试匹配“删除待办” match_del self._delete_pattern.search(message) if match_del: index_str match_del.group(2) try: index int(index_str) - 1 # 用户输入从1开始程序索引从0开始 return await self._delete_todo(user_id, index) except ValueError: return SkillResponse(successFalse, messagef‘{index_str}’不是有效的数字序号。) # 4. 如果都不匹配返回帮助信息 return SkillResponse( successFalse, message我没理解您的指令。您可以这样说\n • 添加待办下午三点开会\n • 查看待办\n • 删除待办 1 ) async def _add_todo(self, user_id: str, content: str) - SkillResponse: if user_id not in self._storage: self._storage[user_id] [] self._storage[user_id].append(content) return SkillResponse( successTrue, messagef已添加待办{content}\n您目前共有 {len(self._storage[user_id])} 项待办。 ) async def _list_todos(self, user_id: str) - SkillResponse: todos self._storage.get(user_id, []) if not todos: return SkillResponse(successTrue, message您目前没有待办事项。) todo_list \n.join([f{i1}. {item} for i, item in enumerate(todos)]) return SkillResponse( successTrue, messagef您的待办事项列表\n{todo_list} ) async def _delete_todo(self, user_id: str, index: int) - SkillResponse: todos self._storage.get(user_id, []) if not todos: return SkillResponse(successFalse, message您目前没有待办事项可删除。) if index 0 or index len(todos): return SkillResponse(successFalse, messagef序号 {index1} 无效请使用‘查看待办’确认序号。) removed_item todos.pop(index) # 如果删除后列表为空可以清理用户键值以节省内存 if not todos: del self._storage[user_id] return SkillResponse( successTrue, messagef已删除待办{removed_item}\n剩余 {len(todos)} 项。 )这个技能实现展示了几个关键点指令解析使用正则表达式进行简单的模式匹配。对于更复杂的自然语言你应该在SkillContext中依赖上游的NLU组件提供的intent和entities字段。状态管理技能内部维护了_storage字典来保存用户数据。这是一个需要特别注意的点在生产环境中技能不应该用内存存储状态而应该连接一个外部持久化存储如Redis、数据库并且要考虑数据隔离不同用户、不同会话。清晰的响应每个分支都返回结构化的SkillResponse包含明确的操作结果和用户友好的消息。3.3 集成大语言模型LLM提升技能智能上面的技能依赖于严格的正则匹配不够灵活。用户如果说“帮我记一下明天要交报告”它就无能为力了。接下来我们改造这个技能引入LLM如OpenAI GPT、国内大模型API来理解用户的自然语言指令。我们将修改can_handle和execute方法加入LLM调用。假设我们有一个LLMClient工具类来封装对大模型API的调用。# todo_skill_llm.py import json from skill_base import BaseSkill, SkillContext, SkillResponse # 假设有一个封装好的LLM客户端 from llm_client import LLMClient class TodoSkillLLM(BaseSkill): def __init__(self, llm_client: LLMClient): self.llm llm_client self._storage {} # 同上建议替换为持久化存储 property def name(self) - str: return todo_manager_llm property def description(self) - str: return 使用LLM理解自然语言的待办事项管理技能。 # 不再依赖关键词触发而是通过LLM判断意图 property def trigger_keywords(self) - list: return [] # 清空完全由can_handle中的LLM判断 def can_handle(self, context: SkillContext) - bool: 使用LLM快速判断消息是否与待办事项相关 # 这是一个轻量级判断可以设计一个简单的提示词 prompt f 用户消息{context.message_text} 请判断这条消息是否与“待办事项”Todo list的管理相关包括添加、查看、删除、查询等操作。 只回答“是”或“否”。 try: # 调用LLM进行快速分类可以设置较低的temperature和token限制以提高速度 response self.llm.chat_completion([{role: user, content: prompt}], max_tokens5) return 是 in response.strip() except Exception as e: # LLM调用失败时可以降级到关键词匹配或直接返回False print(fLLM判断意图失败: {e}) return False async def execute(self, context: SkillContext) - SkillResponse: 使用LLM解析用户指令并执行 # 第一步用LLM将用户指令解析为结构化操作 parse_prompt f 请将以下用户关于待办事项的指令解析为JSON格式。 指令{context.message_text} 输出格式必须严格如下 {{ intent: add_todo|list_todo|delete_todo|unknown, content: 待办内容仅当intent为add_todo时填写, index: 待办序号仅当intent为delete_todo时填写从1开始 }} 如果无法解析或与待办无关intent设为“unknown”。 只输出JSON不要有任何其他解释。 try: parse_response self.llm.chat_completion([{role: user, content: parse_prompt}]) # 清理响应提取JSON部分 json_str parse_response.strip().strip().replace(json\n, ) parsed json.loads(json_str) intent parsed.get(intent, unknown) if intent add_todo: content parsed.get(content, ) if not content: return SkillResponse(successFalse, message请告诉我待办的具体内容。) return await self._add_todo(context.user_id, content) elif intent list_todo: return await self._list_todos(context.user_id) elif intent delete_todo: index parsed.get(index) if index is None: return SkillResponse(successFalse, message请指定要删除的待办序号。) return await self._delete_todo(context.user_id, index - 1) # 转换为0基索引 else: return SkillResponse( successFalse, message我好像没理解您的意思。您可以尝试这样说‘提醒我明天买牛奶’、‘看看我的待办’、‘删除第三个待办’。 ) except json.JSONDecodeError as e: return SkillResponse(successFalse, messagef指令解析失败{e}) except Exception as e: return SkillResponse(successFalse, messagef处理指令时出错{e}) # _add_todo, _list_todos, _delete_todo 方法与之前相同略去...这个版本的技能智能度大大提升。can_handle方法使用LLM进行意图粗筛避免无关消息进入技能。execute方法则利用LLM将自由格式的自然语言转换为结构化的JSON指令从而驱动内部逻辑。这种模式是构建复杂AI技能的核心LLM作为“理解层”传统代码作为“执行层”。实操心得在实际使用LLM时有几点血泪教训提示词工程是关键上面的parse_prompt要求LLM“只输出JSON不要有任何其他解释”至关重要这能极大提高API响应解析的成功率。多轮调试提示词是常态。设置超时与重试LLM API调用可能不稳定一定要在技能外层或LLM客户端内部实现超时和重试机制。成本与延迟考量每次用户消息都调用LLM成本和响应延迟会上升。可以根据场景混合使用高频、简单的指令用规则匹配正则复杂、多变的指令才用LLM。也可以利用can_handle做快速、廉价的过滤。结构化输出是救星尽可能要求LLM输出结构化数据JSON、XML这比解析自由文本稳定得多。最新的模型如GPT-4在遵循输出格式方面表现优异。4. 技能管理器的实现与技能注册有了技能我们需要一个管理器来组织它们。技能管理器是运行时环境与技能之间的桥梁。# skill_manager.py from typing import Dict, List, Type from skill_base import BaseSkill, SkillContext class SkillManager: def __init__(self): self._skills: Dict[str, BaseSkill] {} # name - skill instance def register_skill(self, skill: BaseSkill): 注册一个技能实例 if skill.name in self._skills: raise ValueError(f技能 {skill.name} 已注册。) self._skills[skill.name] skill print(f[技能管理器] 已注册技能: {skill.name} - {skill.description}) def register_from_module(self, module_path: str): 从指定Python模块动态加载并注册所有BaseSkill的子类 # 这里省略了动态导入的细节可以使用importlib # 通常用于插件化架构技能以独立包的形式存在 pass def get_skill_for_context(self, context: SkillContext) - List[BaseSkill]: 根据上下文返回所有能处理该消息的技能按优先级排序 matched_skills [] for skill in self._skills.values(): try: if skill.can_handle(context): matched_skills.append(skill) except Exception as e: print(f技能 {skill.name} 在 can_handle 时出错: {e}) continue # 简单的优先级排序可以基于技能配置的优先级分数或匹配置信度 # 这里我们假设先注册的技能优先级更高或更通用后注册的更具体 return matched_skills async def execute_skill(self, skill_name: str, context: SkillContext) - SkillResponse: 执行指定名称的技能 skill self._skills.get(skill_name) if not skill: return SkillResponse(successFalse, messagef未找到技能 {skill.name}) try: return await skill.execute(context) except Exception as e: print(f执行技能 {skill.name} 时出错: {e}) return SkillResponse(successFalse, message技能执行过程中出现内部错误。) def list_skills(self) - List[Dict]: 列出所有已注册技能的信息 return [{name: s.name, desc: s.description} for s in self._skills.values()]管理器的主要职责是维护技能注册表并根据上下文匹配技能。这里有一个重要的设计点多技能匹配与冲突解决。一条消息可能同时触发多个技能例如消息“明天天气和新闻”可能同时匹配天气技能和新闻技能。管理器需要决定是执行所有匹配技能还是选择一个最优的。常见的策略有优先级排序为每个技能设置静态优先级。置信度打分让can_handle方法返回一个置信度分数选择最高的。人工规则例如关键词完全匹配的优先于模糊匹配的。LLM仲裁在匹配到多个技能时用一句LLM调用决定最合适的那个。在我们的简单实现中get_skill_for_context返回所有匹配的技能由运行时环境决定如何处理例如只执行第一个。5. 与消息运行时环境集成技能和管理器准备好了最后一步是将它们集成到一个具体的消息平台中。我们以一个假设的“Webhook服务器”为例它接收来自钉钉机器人的HTTP请求。# runtime_webhook.py from fastapi import FastAPI, Request from pydantic import BaseModel from skill_manager import SkillManager from skill_base import SkillContext from todo_skill_llm import TodoSkillLLM from llm_client import LLMClient import asyncio app FastAPI() skill_manager SkillManager() # 初始化阶段注册技能 app.on_event(startup) async def startup_event(): llm_client LLMClient(api_keyyour-api-key, modelgpt-3.5-turbo) todo_skill TodoSkillLLM(llm_client) skill_manager.register_skill(todo_skill) # 在这里注册其他技能... print(所有技能注册完毕。) class DingTalkMessage(BaseModel): 钉钉机器人回调消息格式简化版 senderId: str text: dict conversationId: str app.post(/dingtalk/webhook) async def handle_dingtalk_message(request: Request): # 1. 解析请求 dingtalk_msg DingTalkMessage(**(await request.json())) user_id dingtalk_msg.senderId message_text dingtalk_msg.text.get(content, ).strip() session_id dingtalk_msg.conversationId # 2. 构建技能上下文 context SkillContext( user_iduser_id, message_textmessage_text, message_typetext, session_idsession_id ) # 3. 获取匹配的技能 matched_skills skill_manager.get_skill_for_context(context) if not matched_skills: # 没有技能匹配可以返回默认回复或调用一个兜底的“闲聊”技能 return {msgtype: text, text: {content: 抱歉我还没学会处理这个呢。}} # 4. 执行第一个匹配的技能简单策略 # 在实际项目中这里可能会有更复杂的路由逻辑 primary_skill matched_skills[0] response await primary_skill.execute(context) # 5. 将技能响应转换为钉钉所需的格式 if response.success: reply_content response.message else: reply_content f操作未成功{response.message} return { msgtype: text, text: {content: reply_content} } if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)这个运行时环境做了以下几件事协议适配解析钉钉特定的HTTP请求格式将其转化为平台无关的SkillContext。技能路由调用技能管理器找到能处理当前消息的技能。技能执行执行选中的技能并获取结果。响应封装将通用的SkillResponse转换回钉钉机器人要求的JSON格式并返回。这就是openclaw-skill-imsg这类项目所倡导的架构技能开发者只关心业务逻辑Skill.execute运行时开发者只关心协议适配和流程调度两者通过清晰的接口 (SkillContext,SkillResponse) 和组件 (SkillManager) 解耦。6. 进阶话题与生产环境考量将上述Demo代码用于个人学习或原型验证没问题但要投入生产环境还有大量的工程问题需要解决。6.1 技能的生命周期与依赖注入上面的例子中技能在startup_event中硬编码初始化。这不利于管理。一个成熟的系统需要支持动态加载与热更新在不重启服务的情况下添加、移除或更新技能。依赖注入技能可能需要访问数据库连接池、配置中心、其他微服务客户端等共享资源。应该通过一个统一的“技能工厂”或依赖注入容器来创建技能实例并注入这些依赖。配置化技能的参数如LLM的API Key、数据库连接字符串应该从外部配置文件或环境变量读取而不是写在代码里。一个改进的技能管理器可能支持从配置文件中加载技能列表# skills_config.yaml skills: - class: todo_skill_llm.TodoSkillLLM params: llm_client: llm_client # 引用其他已注册的组件 storage_backend: redis_client priority: 10 - class: weather_skill.WeatherSkill params: api_key: ${WEATHER_API_KEY} # 从环境变量读取 cache_ttl: 300 priority: 56.2 上下文管理与多轮对话我们之前的SkillContext只包含了单条消息的信息。要实现多轮对话需要持久化“会话上下文”。例如用户说“添加待办”机器人回复“请告诉我待办内容”用户再说“下午三点开会”机器人需要知道这是对上一个“添加待办”意图的补充。这通常需要会话状态存储在SkillContext中增加一个session对象可以存储任意键值对。技能可以在执行时读写这个状态。状态持久化这个session对象需要被存储到外部数据库如Redis并以(user_id, session_id)为键。每次新消息到来时从存储中恢复上下文。对话超时与清理设置会话超时时间长时间无交互后自动清理状态避免状态膨胀。6.3 错误处理、日志与监控生产级技能必须健壮。全局异常捕获在技能管理器的execute_skill方法外层必须有try...except块捕获所有未处理的异常并返回一个友好的错误响应同时记录详细的错误日志。结构化日志记录每条消息的流入流出、匹配的技能、执行耗时、LLM调用详情等。这对于调试和性能分析至关重要。性能监控为每个技能的execute方法添加计时监控其耗时。对于调用外部API如LLM、天气API的技能更要监控其成功率和延迟。限流与降级如果某个技能依赖的外部服务不可用应有降级策略如返回缓存数据、或提示服务暂时不可用。对于LLM这种可能产生高昂成本的技能应考虑按用户或会话进行调用频率限制。6.4 技能的组合与流水线复杂的任务可能需要多个技能协作完成。例如一个“旅行规划”技能内部可能需要先后调用“天气查询”、“地图导航”、“酒店查询”等多个子技能。这引出了“技能流水线”或“技能编排”的概念。你可以设计一个特殊的OrchestratorSkill它本身也是一个技能。它的execute方法不直接处理业务而是根据一个预定义的流程或动态生成的计划按顺序调用其他技能并将上一个技能的输出作为下一个技能的输入。这需要更复杂的上下文传递和数据共享机制。7. 常见问题与排查技巧实录在实际开发和部署openclaw-skill-imsg这类项目时你会遇到各种各样的问题。下面是我踩过的一些坑和解决方法。问题现象可能原因排查步骤与解决方案技能匹配不上所有消息都走默认回复1.can_handle逻辑有误。2. 技能未正确注册到管理器。3.SkillContext中的message_text为空或格式不对。1. 在can_handle方法内加打印日志确认它被调用且返回值。2. 检查skill_manager.list_skills()输出确认目标技能在列表中。3. 打印context对象确认消息内容已被正确解析。技能执行报错但日志看不到异常异常在异步任务中被吞没。确保在asyncio任务或execute_skill外层有全面的异常捕获和日志记录。使用asyncio.create_task时尤其要注意。LLM技能响应慢导致请求超时1. LLM API网络延迟高。2. 提示词过于复杂导致LLM生成时间长。3. 未设置合理的超时时间。1. 为LLM客户端设置连接和读取超时如10秒。2. 优化提示词减少不必要的上下文。3. 考虑对LLM调用做异步超时控制超时后返回降级响应。多用户同时操作数据互相覆盖或错乱技能内部使用全局变量或类变量存储用户状态未按user_id或session_id隔离。绝对禁止在技能类中使用self._storage {}这样的类属性存储用户数据。必须使用外部存储如Redis并以f”todo:{user_id}”这样的键来区分。技能在本地运行正常部署后无法加载1. 依赖包版本不一致。2. 动态导入路径问题。3. 环境变量未设置。1. 使用pip freeze requirements.txt和虚拟环境确保一致性。2. 检查部署环境的PYTHONPATH和当前工作目录。3. 使用os.getenv并设置合理的默认值部署时确认环境变量已注入。中文关键词匹配失败字符串匹配时未考虑编码或空格问题。1. 在匹配前对message_text进行.strip()和可能的全角/半角转换。2. 使用keyword in message_text是安全的但要注意子串误匹配如“天气”匹配“今天天气不错”是OK的但“天”也会匹配“今天”这可能需要更精确的正则或分词。我个人在实际操作中的体会是开发AI技能库项目三分在算法和模型七分在工程和架构。一开始很容易沉迷于用LLM做出炫酷的对话效果但很快就会被并发、状态管理、错误处理、部署运维这些“脏活累活”拖垮。因此在项目早期就确立清晰的分层架构协议层、运行时层、技能层、定义稳固的数据接口Context/Response、并规划好技能的开发规范如如何注入依赖、如何记录日志比急于实现功能重要得多。这能保证项目在规模扩大、技能增多时依然保持可维护性和可扩展性。先从一个小而美的技能开始跑通整个流程再逐步迭代增加复杂度和规模是更稳妥的路径。