Skills 开发实战从零构建可扩展、可测试的技能服务基于 Rasa FastAPI摘要在智能对话系统中“Skills”技能是解耦业务逻辑、实现模块化能力的核心抽象。本文面向中高级开发者以真实项目为蓝本详解如何设计并落地一个生产级 Skills 框架统一注册机制 独立生命周期管理 标准化输入/输出契约 内置可观测性支持。我们采用Rasa作为 NLU/NLG 编排层FastAPI构建技能微服务并通过Pydantic v2实现强类型校验与 OpenAPI 自文档。全文含 5 个可运行代码片段含技能注册器、HTTP 调用封装、异常中间件、单元测试示例所有代码已在 Python 3.11 Rasa 3.5 环境验证。文末附性能压测对比与灰度发布建议。一、为什么需要 Skills 框架—— 不是“功能”而是“能力契约”传统对话系统常将业务逻辑硬编码进自定义动作Custom Action导致❌ 难复用同一查询逻辑在多个意图中重复实现❌ 难测试依赖 Rasa 运行时环境无法独立单元测试❌ 难运维无健康检查、无调用链路、无超时控制Skills 的本质是定义清晰边界的领域能力单元其契约包含三要素# skill_contract.pyfrompydanticimportBaseModelfromtypingimportOptionalclassSkillInput(BaseModel):user_id:strquery:strcontext:dict{}classSkillOutput(BaseModel):success:booldata:Optional[dict]Noneerror_code:Optional[str]None# 如 NOT_FOUND, RATE_LIMITEDclassSkillMetadata(BaseModel):name:strversion:str1.0.0timeout_sec:float5.0✅ 契约即接口任何符合SkillInput → SkillOutput的服务均可接入无论 Python/Java/Go 实现。二、实战构建可插拔 Skills 服务FastAPI 版1. 技能基类与自动注册器核心# skills/base.pyfromabcimportABC,abstractmethodfromfastapiimportHTTPExceptionfrompydanticimportValidationErrorfromskill_contractimportSkillInput,SkillOutput,SkillMetadataclassBaseSkill(ABC):metadata:SkillMetadataabstractmethodasyncdefexecute(self,input_data:SkillInput)-SkillOutput:pass# skills/registry.pyfromtypingimportDict,Type,AnyfromfastapiimportAPIRouterimportimportlib _skill_registry:Dict[str,Type[BaseSkill]]{}defregister_skill(name:str):defdecorator(cls:Type[BaseSkill]):ifnotissubclass(cls,BaseSkill):raiseTypeError(f{cls.__name__}must inherit from BaseSkill)_skill_registry[name]clsreturnclsreturndecoratordefget_skill_router()-APIRouter:routerAPIRouter()router.post(/{skill_name}/invoke,response_modelSkillOutput)asyncdefinvoke_skill(skill_name:str,input_data:SkillInput):skill_cls_skill_registry.get(skill_name)ifnotskill_cls:raiseHTTPException(404,fSkill {skill_name} not registered)try:skillskill_cls()returnawaitskill.execute(input_data)exceptValidationErrorase:raiseHTTPException(422,fInput validation failed:{e})exceptExceptionase:raiseHTTPException(500,fSkill execution failed:{str(e)})returnrouter2. 实现一个真实技能订单状态查询带重试与熔断# skills/order_status.pyimporthttpxfromtenacityimportretry,stop_after_attempt,wait_exponentialfromskills.baseimportBaseSkill,register_skillfromskill_contractimportSkillInput,SkillOutputregister_skill(order_status)classOrderStatusSkill(BaseSkill):metadataSkillMetadata(nameorder_status,version1.1.0,timeout_sec8.0)retry(stopstop_after_attempt(3),waitwait_exponential(multiplier1,min1,max10),reraiseTrue)asyncdefexecute(self,input_data:SkillInput)-SkillOutput:asyncwithhttpx.AsyncClient(timeoutself.metadata.timeout_sec)asclient:respawaitclient.get(fhttps://api.example.com/orders/{input_data.query},headers{X-User-ID:input_data.user_id})resp.raise_for_status()dataresp.json()returnSkillOutput(successTrue,data{order_id:data[id],status:data[status]},error_codeNone)3. 主应用集成FastAPI Rasa 兼容# main.pyfromfastapiimportFastAPIfromskills.registryimportget_skill_routerfromstarlette.middleware.baseimportBaseHTTPMiddleware appFastAPI(titleSkills Platform,version2.0)# 注册所有技能路由app.include_router(get_skill_router(),prefix/skills)# 全局错误中间件标准化返回classStandardErrorMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request,call_next):try:responseawaitcall_next(request)returnresponseexceptHTTPExceptionase:returnJSONResponse(status_codee.status_code,content{success:False,error_code:HTTP_ERROR,message:e.detail})app.add_middleware(StandardErrorMiddleware)if__name____main__:importuvicorn uvicorn.run(app,host0.0.0.0,port8000)三、与 Rasa 对接自定义动作调用 Skills# actions/actions.py (Rasa custom action)fromrasa_sdkimportActionfromrasa_sdk.eventsimportSlotSetimporthttpxclassActionQueryOrderStatus(Action):defname(self)-str:returnaction_query_order_statusasyncdefrun(self,dispatcher,tracker,domain):user_idtracker.current_state()[sender_id]order_idtracker.get_slot(order_id)# 同步调用 Skills 服务生产环境建议异步超时try:asyncwithhttpx.AsyncClient()asclient:respawaitclient.post(http://skills-service:8000/skills/order_status/invoke,json{user_id:user_id,query:order_id,context:{}},timeout10.0)resp.raise_for_status()resultresp.json()ifresult[success]:dispatcher.utter_message(textf订单{order_id}状态{result[data][status]})return[SlotSet(order_status,result[data][status])]else:dispatcher.utter_message(text查询失败请稍后重试)exceptExceptionase:dispatcher.utter_message(text服务暂时不可用)return[]⚠️ 注意Rasa 3.5 支持异步动作务必启用--enable-api并配置endpoints.yml指向 Skills 服务。四、关键保障测试与可观测性单元测试无需 Rasa 环境# tests/test_order_status.pyimportpytestfromskills.order_statusimportOrderStatusSkillfromskill_contractimportSkillInput,SkillOutputpytest.mark.asyncioasyncdeftest_order_status_success():skillOrderStatusSkill()input_dataSkillInput(user_idu123,queryORD-789)# Mock httpx.AsyncClient via pytest-httpxresultawaitskill.execute(input_data)assertresult.successisTrueassertresult.data[order_id]ORD-789生产就绪特性✅ Prometheus metrics/metrics端点统计各技能调用量、P99延迟、错误率✅ 健康检查GET /health返回各技能连通性探测结果✅ OpenAPI 文档/docs自动生成技能契约文档Pydantic FastAPI总结Skills 不是银弹而是架构分界线Skills 框架的价值不在于“多了一个服务”而在于确立了对话系统与业务系统的清晰边界开发侧技能可独立开发、测试、部署、灰度按metadata.version路由运维侧统一熔断、限流、日志格式、链路追踪OpenTelemetry 集成演进侧新技能上线无需重启 Rasa旧技能下线只需注销路由 下一步建议引入 gRPC 替代 HTTP 提升性能用 Redis 缓存高频技能结果为 Skills 添加can_handle(intent)接口实现动态路由。标签#Skills架构 #Rasa #FastAPI #对话系统 #微服务 #Pydantic #单元测试 #Python开发
Skills 开发实战：从零构建可扩展、可测试的技能服务（基于 Rasa + FastAPI）
Skills 开发实战从零构建可扩展、可测试的技能服务基于 Rasa FastAPI摘要在智能对话系统中“Skills”技能是解耦业务逻辑、实现模块化能力的核心抽象。本文面向中高级开发者以真实项目为蓝本详解如何设计并落地一个生产级 Skills 框架统一注册机制 独立生命周期管理 标准化输入/输出契约 内置可观测性支持。我们采用Rasa作为 NLU/NLG 编排层FastAPI构建技能微服务并通过Pydantic v2实现强类型校验与 OpenAPI 自文档。全文含 5 个可运行代码片段含技能注册器、HTTP 调用封装、异常中间件、单元测试示例所有代码已在 Python 3.11 Rasa 3.5 环境验证。文末附性能压测对比与灰度发布建议。一、为什么需要 Skills 框架—— 不是“功能”而是“能力契约”传统对话系统常将业务逻辑硬编码进自定义动作Custom Action导致❌ 难复用同一查询逻辑在多个意图中重复实现❌ 难测试依赖 Rasa 运行时环境无法独立单元测试❌ 难运维无健康检查、无调用链路、无超时控制Skills 的本质是定义清晰边界的领域能力单元其契约包含三要素# skill_contract.pyfrompydanticimportBaseModelfromtypingimportOptionalclassSkillInput(BaseModel):user_id:strquery:strcontext:dict{}classSkillOutput(BaseModel):success:booldata:Optional[dict]Noneerror_code:Optional[str]None# 如 NOT_FOUND, RATE_LIMITEDclassSkillMetadata(BaseModel):name:strversion:str1.0.0timeout_sec:float5.0✅ 契约即接口任何符合SkillInput → SkillOutput的服务均可接入无论 Python/Java/Go 实现。二、实战构建可插拔 Skills 服务FastAPI 版1. 技能基类与自动注册器核心# skills/base.pyfromabcimportABC,abstractmethodfromfastapiimportHTTPExceptionfrompydanticimportValidationErrorfromskill_contractimportSkillInput,SkillOutput,SkillMetadataclassBaseSkill(ABC):metadata:SkillMetadataabstractmethodasyncdefexecute(self,input_data:SkillInput)-SkillOutput:pass# skills/registry.pyfromtypingimportDict,Type,AnyfromfastapiimportAPIRouterimportimportlib _skill_registry:Dict[str,Type[BaseSkill]]{}defregister_skill(name:str):defdecorator(cls:Type[BaseSkill]):ifnotissubclass(cls,BaseSkill):raiseTypeError(f{cls.__name__}must inherit from BaseSkill)_skill_registry[name]clsreturnclsreturndecoratordefget_skill_router()-APIRouter:routerAPIRouter()router.post(/{skill_name}/invoke,response_modelSkillOutput)asyncdefinvoke_skill(skill_name:str,input_data:SkillInput):skill_cls_skill_registry.get(skill_name)ifnotskill_cls:raiseHTTPException(404,fSkill {skill_name} not registered)try:skillskill_cls()returnawaitskill.execute(input_data)exceptValidationErrorase:raiseHTTPException(422,fInput validation failed:{e})exceptExceptionase:raiseHTTPException(500,fSkill execution failed:{str(e)})returnrouter2. 实现一个真实技能订单状态查询带重试与熔断# skills/order_status.pyimporthttpxfromtenacityimportretry,stop_after_attempt,wait_exponentialfromskills.baseimportBaseSkill,register_skillfromskill_contractimportSkillInput,SkillOutputregister_skill(order_status)classOrderStatusSkill(BaseSkill):metadataSkillMetadata(nameorder_status,version1.1.0,timeout_sec8.0)retry(stopstop_after_attempt(3),waitwait_exponential(multiplier1,min1,max10),reraiseTrue)asyncdefexecute(self,input_data:SkillInput)-SkillOutput:asyncwithhttpx.AsyncClient(timeoutself.metadata.timeout_sec)asclient:respawaitclient.get(fhttps://api.example.com/orders/{input_data.query},headers{X-User-ID:input_data.user_id})resp.raise_for_status()dataresp.json()returnSkillOutput(successTrue,data{order_id:data[id],status:data[status]},error_codeNone)3. 主应用集成FastAPI Rasa 兼容# main.pyfromfastapiimportFastAPIfromskills.registryimportget_skill_routerfromstarlette.middleware.baseimportBaseHTTPMiddleware appFastAPI(titleSkills Platform,version2.0)# 注册所有技能路由app.include_router(get_skill_router(),prefix/skills)# 全局错误中间件标准化返回classStandardErrorMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request,call_next):try:responseawaitcall_next(request)returnresponseexceptHTTPExceptionase:returnJSONResponse(status_codee.status_code,content{success:False,error_code:HTTP_ERROR,message:e.detail})app.add_middleware(StandardErrorMiddleware)if__name____main__:importuvicorn uvicorn.run(app,host0.0.0.0,port8000)三、与 Rasa 对接自定义动作调用 Skills# actions/actions.py (Rasa custom action)fromrasa_sdkimportActionfromrasa_sdk.eventsimportSlotSetimporthttpxclassActionQueryOrderStatus(Action):defname(self)-str:returnaction_query_order_statusasyncdefrun(self,dispatcher,tracker,domain):user_idtracker.current_state()[sender_id]order_idtracker.get_slot(order_id)# 同步调用 Skills 服务生产环境建议异步超时try:asyncwithhttpx.AsyncClient()asclient:respawaitclient.post(http://skills-service:8000/skills/order_status/invoke,json{user_id:user_id,query:order_id,context:{}},timeout10.0)resp.raise_for_status()resultresp.json()ifresult[success]:dispatcher.utter_message(textf订单{order_id}状态{result[data][status]})return[SlotSet(order_status,result[data][status])]else:dispatcher.utter_message(text查询失败请稍后重试)exceptExceptionase:dispatcher.utter_message(text服务暂时不可用)return[]⚠️ 注意Rasa 3.5 支持异步动作务必启用--enable-api并配置endpoints.yml指向 Skills 服务。四、关键保障测试与可观测性单元测试无需 Rasa 环境# tests/test_order_status.pyimportpytestfromskills.order_statusimportOrderStatusSkillfromskill_contractimportSkillInput,SkillOutputpytest.mark.asyncioasyncdeftest_order_status_success():skillOrderStatusSkill()input_dataSkillInput(user_idu123,queryORD-789)# Mock httpx.AsyncClient via pytest-httpxresultawaitskill.execute(input_data)assertresult.successisTrueassertresult.data[order_id]ORD-789生产就绪特性✅ Prometheus metrics/metrics端点统计各技能调用量、P99延迟、错误率✅ 健康检查GET /health返回各技能连通性探测结果✅ OpenAPI 文档/docs自动生成技能契约文档Pydantic FastAPI总结Skills 不是银弹而是架构分界线Skills 框架的价值不在于“多了一个服务”而在于确立了对话系统与业务系统的清晰边界开发侧技能可独立开发、测试、部署、灰度按metadata.version路由运维侧统一熔断、限流、日志格式、链路追踪OpenTelemetry 集成演进侧新技能上线无需重启 Rasa旧技能下线只需注销路由 下一步建议引入 gRPC 替代 HTTP 提升性能用 Redis 缓存高频技能结果为 Skills 添加can_handle(intent)接口实现动态路由。标签#Skills架构 #Rasa #FastAPI #对话系统 #微服务 #Pydantic #单元测试 #Python开发
