1. “Skill”不是功能模块,而是Agent世界的“原子操作单元”
最近在好几个技术群里看到新人发问:“我用Coze搭了个天气Bot,加了‘查天气’‘设提醒’‘讲笑话’三个按钮,这算不算实现了三个Skill?”——答案是否定的。这种把界面按钮、API调用、甚至一段Python脚本直接冠以“Skill”之名的做法,正在快速污染整个Agent开发圈的认知基线。我带过六支Agent产品团队,从金融风控Agent到工业设备巡检Agent,踩过最深的坑,恰恰就始于对“Skill”二字的轻率定义。
Skill不是功能列表里的一个条目,不是前端UI上的一块按钮区域,更不是后端服务里一个暴露出来的HTTP接口。它是一套有明确定义边界、可独立验证、具备语义完整性、能被Agent运行时统一调度的最小行为单元。你可以把它理解成操作系统里的“系统调用”(syscall):open()、read()、write()这些不是随便起的名字,它们代表内核对外暴露的、经过严格契约约束的原子能力。同理,search_web(query: str) → List[Result]是Skill,call_weather_api(city: str) → dict是Skill,但show_weather_card()就不是——后者是UI渲染逻辑,属于Presentation Layer,和Skill所在的Execution Layer根本不在同一抽象层级。
为什么这个区分如此关键?因为一旦混淆,工程实现就会立刻崩塌。我们曾在一个政务问答Agent项目中,把“生成PDF报告”封装成一个Skill,结果上线后发现:当用户连续发起5次请求时,PDF生成服务因临时文件未清理而OOM;当并发量超过80QPS时,字体嵌入失败导致文档乱码;更致命的是,该“Skill”内部硬编码了本地路径/tmp/report/,导致在K8s集群多副本部署时,不同Pod写入冲突,报告内容相互覆盖。问题根源不是代码写得差,而是从设计第一天起,就把一个依赖强状态、强IO、强环境耦合的复合操作,错误地当成了无状态、可组合、可重入的Skill。
真正的Skill必须满足四个刚性条件:
- 契约明确性:输入参数类型、取值范围、必填项;输出结构、成功/失败标识、错误码分类,全部通过TypeScript Interface或OpenAPI Schema显式声明,不能靠文档“约定俗成”。
- 副作用隔离性:执行过程不修改全局状态,不依赖外部文件系统路径,不持有长连接句柄。所有外部依赖(数据库、HTTP Client)必须通过Dependency Injection注入,且在测试中可被Mock。
- 幂等可重试性:同一输入在相同上下文(如Agent Session ID)下重复执行,应返回相同结果或明确的幂等响应(如
{status: "already_done", result_id: "xxx"}),而非抛出“记录已存在”异常。 - 可观测可审计性:每次调用必须生成结构化日志(含Skill Name、Input Hash、Execution Duration、Exit Code),并支持按Session ID、User ID、Skill Name三维度实时聚合分析。
这四条不是理想主义的教条,而是我们在生产环境用27次严重事故换来的血泪清单。比如某次电商Agent的“下单”Skill因未实现幂等性,在网络抖动重试时导致用户被扣款两次;某次教育Agent的“生成错题解析”Skill因未做副作用隔离,将中间缓存写入共享Redis,致使A学生看到B学生的解题思路。这些故障的根因,90%以上都指向同一个源头:把“能跑通”的代码,当成了“可信赖”的Skill。
提示:判断一个操作是否够格成为Skill,最朴素的测试法是——把它从当前Agent框架中完全剥离,单独部署为一个独立微服务(哪怕只是本地HTTP Server),看它能否仅凭输入JSON,稳定返回符合Schema的输出JSON。如果需要额外传入Session Context、User Profile、Config Map才能工作,那它大概率还不是Skill,而是某个Skill的“执行器”(Executor)。
2. Skill的本质是“意图-动作”的精准映射,而非代码片段的简单打包
很多开发者构建Skill时,习惯性打开IDE,新建一个Python文件,写个def get_stock_price(ticker: str),再加个@skill装饰器,就宣告完成。这种做法看似高效,实则埋下了巨大的语义鸿沟。Skill的核心价值,从来不在代码本身,而在于它如何将人类自然语言意图,无损、无歧义、可追溯地翻译为机器可执行的动作。这个翻译过程,才是Skill工程化的真正战场。
我们以一个真实案例切入:某银行智能投顾Agent收到用户指令“帮我把活期账户里30%的钱转到余额宝”。表面看,这只是一个转账操作,但拆解其背后意图链,会发现至少包含五个不可跳过的语义节点:
- 账户识别:用户说的“活期账户”指代哪个?是工资卡I类户,还是网银绑定的II类电子账户?需结合用户KYC数据与账户标签体系动态解析;
- 资产计算:30%是基于当前可用余额,还是基于昨日收盘后总资产?是否排除冻结资金?需调用实时资金引擎获取精确快照;
- 目标产品匹配:“余额宝”在银行系统内对应的是“天弘基金-增利宝货币A”,但用户可能用“零钱通”“理财通”等别名,需建立标准化产品ID与别名词典;
- 合规校验:单日转账是否超监管限额?用户风险评级是否匹配该理财产品?需同步调用反洗钱引擎与适当性管理服务;
- 执行确认:转账前必须向用户展示预估收益、到账时间、手续费,并获得显式授权(非默认勾选),否则违反《金融消费者权益保护实施办法》。
如果把上述全部逻辑塞进一个叫transfer_to_money_market()的Skill里,它就成了典型的“上帝函数”(God Function):职责爆炸、测试困难、无法复用、难以审计。正确的工程实践,是将其拆解为五个正交的Skill:
resolve_account(intent: str) → AccountIdcalculate_available_balance(account_id: str) → Decimalmap_product_alias(alias: str) → FundProductIdvalidate_compliance(account_id: str, fund_id: str, amount: Decimal) → ComplianceResultexecute_transfer(account_id: str, fund_id: str, amount: Decimal) → TransferReceipt
这五个Skill之间,通过标准Context对象传递数据,每个Skill只负责解决一个原子语义问题。当用户指令变更(如“转5万元到招行朝朝宝”),只需调整map_product_alias的映射规则,其余Skill完全无需修改。这才是Skill作为“意图-动作映射单元”的威力所在——它让Agent的决策链路变得像乐高积木一样,可插拔、可替换、可审计。
那么,如何确保这种映射不丢失语义?关键在于引入Skill Schema + Intent Parser双轨验证机制。我们强制要求每个Skill必须配套一个.skill.yaml描述文件,例如transfer_to_money_market.skill.yaml:
name: transfer_to_money_market version: "1.2.0" description: "将指定账户资金转入货币基金产品,执行前自动完成合规校验" input_schema: type: object properties: source_account_id: type: string description: "源账户唯一标识符,由resolve_account Skill返回" pattern: "^ACC_[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" target_fund_id: type: string description: "目标货币基金产品ID,由map_product_alias Skill返回" pattern: "^FUND_[A-Z]{3,6}_\\d+$" amount: type: number minimum: 0.01 maximum: 10000000.00 multipleOf: 0.01 output_schema: type: object properties: receipt_id: type: string description: "转账回执单号,可用于后续查询" expected_arrival: type: string format: date-time description: "预计到账时间(ISO 8601格式)" fee: type: number description: "本次转账手续费(元)"这个YAML文件不是文档,而是编译期契约。Agent框架在加载Skill时,会用JSON Schema Validator严格校验输入输出,任何类型不符、字段缺失、数值越界,都会在调用前抛出SkillContractViolationError,而非让错误流入下游服务。同时,Intent Parser(如基于LLM的Router)在分发指令前,会根据此Schema反向生成Prompt模板:“请从用户输入中提取source_account_id、target_fund_id、amount三个字段,若任一字段缺失或模糊,请返回NOT_ENOUGH_INFO”。这样,从用户输入到Skill执行,全程处于契约约束之下,彻底杜绝“传参错位”“字段误解”这类低级但高频的故障。
注意:不要试图用LLM直接解析原始用户输入生成Skill调用参数。我们做过AB测试:纯LLM解析的准确率在复杂场景下仅为68.3%,而“LLM粗筛+Schema精校验”组合方案可达99.2%。前者把信任交给黑盒,后者把控制权握在自己手中。
3. SKILL.md不是README,而是Skill的“数字身份证”与协作协议书
在开源社区和内部项目中,我见过太多名为SKILL.md的文件,内容不过是几行代码片段加一句“调用此函数即可”。这种文档,本质上是对Skill工程化精神的背叛。真正的SKILL.md,绝不是给开发者看的“怎么用”,而是给Agent运行时、监控系统、审计员、安全团队、甚至未来接手的同事看的“它到底是什么、能做什么、不能做什么、出了问题怎么查”。
一份合格的SKILL.md,必须包含七个不可省略的模块,缺一不可。我们以一个真实的generate_tax_reportSkill为例,逐项说明其内容构成与设计逻辑:
3.1 基础元信息(Machine-Readable Header)
--- name: generate_tax_report version: 2.1.0 author: finance-team@company.com license: Proprietary last_updated: 2024-05-12 schema_version: 1.0 ---这不是装饰性YAML。schema_version字段决定了Agent框架如何解析后续内容;license字段在跨团队调用时触发合规检查(如开源组件禁止调用闭源Skill);last_updated与Git Commit Hash绑定,用于灰度发布时精准定位版本。
3.2 语义定义(Human-Readable Intent Mapping)
## 语义定义 本Skill响应以下用户意图: - “帮我导出上季度个税申报表” - “生成2024年Q1的纳税汇总报告” - “我要查看公司代扣代缴明细” **不响应以下意图**: - “怎么报税?”(这是FAQ类问答,应路由至Knowledge Base Skill) - “个税起征点是多少?”(这是政策查询,应路由至Regulation API Skill) - “把报表发到邮箱”(发送动作需由Notification Skill执行,本Skill只生成PDF二进制流)这里用正反例明确划清Skill的语义边界。我们曾因缺少“不响应”条款,在税务Agent中误将政策咨询路由至此Skill,导致LLM反复尝试生成不存在的政策文本,最终触发Rate Limit熔断。
3.3 输入输出契约(Code-First Schema)
## 输入输出契约 ### 输入参数(JSON Schema) ```json { "type": "object", "properties": { "user_id": {"type": "string"}, "report_period": { "type": "string", "enum": ["Q1-2024", "Q2-2024", "H1-2024", "YEAR-2023"] }, "format": {"type": "string", "enum": ["pdf", "xlsx"]} }, "required": ["user_id", "report_period"] }输出结构(Protobuf Message Definition)
message TaxReportResponse { bytes report_data = 1; // PDF or XLSX binary content string report_id = 2; // UUID for audit trail int32 status_code = 3; // 0=success, 1=data_missing, 2=permission_denied string error_message = 4; }契约必须同时提供JSON Schema(供HTTP调用)和Protobuf(供gRPC调用),确保跨协议一致性。status_code枚举值必须与公司统一错误码中心对齐,避免各Skill自定义"error": "not found"这类模糊表述。
3.4 执行约束(Operational Boundaries)
## 执行约束 - **时效性**:单次执行耗时 ≤ 8.5秒(P95),超时自动终止并返回`status_code=4`(TIMEOUT) - **数据范围**:仅处理用户本人及直系亲属(配偶、未成年子女)的纳税数据,需通过`auth_service.verify_relationship()`二次鉴权 - **资源消耗**:内存占用 ≤ 1.2GB,CPU使用率 ≤ 70%(由K8s Resource Limit强制保障) - **重试策略**:网络超时可重试2次,数据源不可用(如税务API返回503)则立即失败,不重试这些约束不是性能指标,而是SLO(Service Level Objective)承诺。监控系统会实时采集execution_duration_ms、memory_usage_mb等指标,一旦连续3分钟违反约束,自动触发告警并降级至备用Skill。
3.5 安全与合规(Compliance Artifacts)
## 安全与合规 - **数据脱敏**:输出PDF中所有身份证号、银行卡号、手机号均按`****-****-****-1234`格式掩码,掩码规则由`data_masking_engine`统一执行 - **审计日志**:每次调用生成两条日志: - `audit.log`: `{"skill":"generate_tax_report","user_id":"U123","report_period":"Q1-2024","ip":"10.1.2.3","timestamp":"2024-05-12T08:30:45Z"}` - `security.log`: `{"event":"PII_ACCESS","fields":["id_card","bank_account"],"masking_applied":true,"timestamp":"..."}` - **合规认证**:已通过等保三级测评(证书编号:DJBZ-2024-XXXXX),支持GDPR Right-to-Erasure请求没有这一节,Skill在金融、医疗等强监管领域根本无法上线。security.log专为安全团队设计,确保PII(Personally Identifiable Information)访问行为100%可追溯。
3.6 故障诊断(Troubleshooting Guide)
## 故障诊断 | 现象 | 可能原因 | 排查命令 | 解决方案 | |------|----------|----------|----------| | `status_code=1` | 用户纳税数据未同步至数仓 | `curl -X GET "http://data-warehouse/api/v1/users/U123/tax?period=Q1-2024"` | 触发手动数据同步Job `sync_tax_data --user U123 --period Q1-2024` | | `status_code=2` | 用户权限不足(非本人/亲属) | `curl -X GET "http://auth-service/api/v1/relationship/U123"` | 检查`relationship_type`字段是否为`SELF`、`SPOUSE`或`CHILD` | | `status_code=4` | PDF生成服务OOM | `kubectl top pods -n tax-reporting` | 扩容`pdf-generator` Deployment至3副本 |这不是运维手册,而是给一线工程师的“秒级定位指南”。表格中每条“排查命令”都经过实测,确保复制粘贴即可执行,避免出现kubectl get pods -n xxx这种无效指令。
3.7 版本演进(Version History)
## 版本演进 ### v2.1.0 (2024-05-12) - 新增`format=xlsx`支持,输出结构化Excel(列:收入类型、金额、税率、已缴税额) - 将`status_code=3`(INTERNAL_ERROR)拆分为`4`(TIMEOUT)和`5`(SERVICE_UNAVAILABLE),提升错误可读性 - 移除对旧版税务API的兼容,强制升级至v3.2 ### v1.8.0 (2024-02-20) - 首次发布,支持PDF格式,基础个税计算逻辑 - 依赖`tax-calculation-lib@1.4.0`,已通过FIPS 140-2加密认证版本历史不是流水账,而是影响面评估清单。v2.1.0的“移除旧API兼容”意味着所有调用方必须同步升级,此条目会自动触发CI/CD Pipeline中的兼容性检查。
提示:
SKILL.md必须纳入Git Hooks校验。我们配置了pre-commit hook,强制要求:
name字段必须与文件名一致(generate_tax_report.skill.md→name: generate_tax_report);version字段必须遵循SemVer 2.0规范;input_schema必须能被jsonschema库成功加载。
任何一条不满足,commit直接被拒绝。文档即代码,失信于文档,就是失信于整个系统。
4. Skill工程化的四大反模式:你正在踩中的那些坑
在Review超过1200个内部Skill实现后,我总结出四个高频、隐蔽、且后果严重的反模式。它们不像语法错误那样立刻报红,而是在系统规模扩大、流量峰值到来、合规审计启动时,才集中爆发。识别并规避这些反模式,是Skill工程化落地的生死线。
4.1 反模式一:Skill即胶水层(The Glue-Anti-Pattern)
典型症状:Skill代码里充斥着requests.get()、subprocess.run()、open('/path/to/config.json'),核心逻辑只有3行,其余全是适配不同API的转换代码。开发者美其名曰“统一接入层”,实则是把Skill变成了技术债黑洞。
为什么危险?
- 可维护性归零:当上游API变更(如微信支付接口升级v3),你需要修改所有调用它的Skill,而非集中更新一个Client SDK;
- 可观测性断裂:
requests.get()的耗时、错误码、重试次数,无法被统一Metrics Collector捕获,监控大盘上只剩一个模糊的skill_execution_duration; - 安全漏洞温床:硬编码的API Key、未校验的SSL证书、未过滤的用户输入拼接URL,全部在Skill内部裸奔。
正确解法:Skill必须只依赖抽象接口,而非具体实现。
我们强制推行“三层依赖架构”:
- Skill层:只调用
payment_gateway.charge(amount: Decimal, currency: str)这样的抽象方法; - Adapter层:由专门团队维护
WechatPayAdapter、AlipayAdapter、StripeAdapter,各自实现charge()方法,并封装重试、熔断、日志; - Client SDK层:提供标准化HTTP Client(如
httpx.AsyncClientwithRetryMiddleware),所有Adapter复用同一套网络栈。
这样,当微信支付升级,只需更新WechatPayAdapter,所有调用它的Skill自动受益。我们曾用此架构,在2小时内完成全站17个支付相关Skill的v3接口迁移,零 downtime。
4.2 反模式二:状态外溢(The State-Leak-Anti-Pattern)
典型症状:Skill函数内部创建全局变量、缓存字典、单例连接池,甚至直接操作threading.local()存储用户上下文。开发者认为“反正只在本Skill里用,没问题”。
为什么危险?
- 并发安全崩溃:在异步框架(如FastAPI)中,
threading.local()无法隔离协程,A用户的Session数据被B用户意外读取; - 内存泄漏雪球:缓存字典随请求累积,GC无法回收,K8s Pod内存持续增长直至OOM Kill;
- 分布式失效:单机缓存无法在多副本间同步,导致同一用户在不同Pod上看到不一致的结果。
正确解法:Skill必须是纯函数(Pure Function)或显式状态管理。
- 纯函数路径:输入→计算→输出,零副作用。适用于90%的Skill,如
calculate_compound_interest(principal, rate, years); - 显式状态路径:若必须维护状态(如对话状态机),则通过
context: SkillContext参数传入,且SkillContext必须实现__slots__限制属性,禁止动态添加字段。SkillContext由Agent Runtime统一创建、注入、销毁,确保生命周期可控。
我们用Pydantic V2定义SkillContext:
class SkillContext(BaseModel): session_id: str user_id: str timestamp: datetime # 显式声明允许的状态字段,禁止动态扩展 __slots__ = ("session_id", "user_id", "timestamp", "_cache") def set_cache(self, key: str, value: Any) -> None: if not hasattr(self, "_cache"): self._cache = {} self._cache[key] = value def get_cache(self, key: str) -> Any: return self._cache.get(key)任何试图ctx.new_field = "boom"的操作,都会触发AttributeError,在开发阶段就暴露问题。
4.3 反模式三:错误处理即静默吞咽(The Silent-Swallow-Anti-Pattern)
典型症状:Skill代码里大量try...except Exception as e: logger.error(e); return None,或者更糟——except:后面直接pass。开发者觉得“别让错误崩掉整个Agent”。
为什么危险?
- 故障不可见:监控系统收不到错误指标,告警静默,直到用户投诉才发觉;
- 数据腐化:
return None被上游Skill当作有效结果,导致空报告、零余额、错误跳转; - 根因难追溯:日志只有
Exception occurred,没有堆栈、没有输入、没有上下文,排查耗时翻倍。
正确解法:错误必须分类、可量化、可追溯。
我们定义三类错误:
- UserError(4xx类):用户输入非法,如
amount < 0,返回{"status": "USER_ERROR", "code": "INVALID_AMOUNT", "message": "金额不能为负数"}; - SystemError(5xx类):服务暂时不可用,如DB连接超时,返回
{"status": "SYSTEM_ERROR", "code": "DB_TIMEOUT", "retry_after": 1000},并自动触发重试; - FatalError(Panic类):程序逻辑崩溃,如
None被当作int相加,必须raise原始异常,由Agent Runtime捕获并记录完整堆栈。
关键创新点在于:所有错误响应,必须携带trace_id和input_hash。当用户投诉“生成报告失败”,客服只需提供trace_id,运维就能在ELK中秒级检索到该次调用的完整输入、输出、日志、堆栈,无需用户复述操作步骤。
4.4 反模式四:测试即Hello World(The Hello-World-Test-Anti-Pattern)
典型症状:Skill目录下有个test_basic.py,里面只有一行assert skill_function("test") == "success"。开发者认为“能跑通就算测试覆盖”。
为什么危险?
- 边界场景全裸奔:
amount=0、amount=999999999.99、user_id="../../../etc/passwd"等边界值从未被验证; - 契约漂移无感知:Schema变更后,旧测试用例仍通过,因为没校验输出结构;
- 性能瓶颈被掩盖:单次调用10ms,但100并发时飙升至5s,测试从未模拟真实负载。
正确解法:测试必须覆盖契约、边界、性能、混沌四维度。
我们要求每个Skill必须有四个测试套件:
- ContractTest:用Pydantic Model加载
input_schema和output_schema,对任意输入生成随机合法/非法数据,验证Schema校验逻辑; - BoundaryTest:针对每个数值型参数,测试
min-1、min、max、max+1、null、NaN; - LoadTest:用Locust模拟100并发,测量P95延迟、错误率、内存增长曲线,阈值写死在
pyproject.toml中; - ChaosTest:在测试环境注入故障——随机Kill DB Pod、注入500ms网络延迟、篡改ConfigMap,验证Skill的降级与熔断行为。
一次真实的ChaosTest救了我们:在模拟Redis宕机时,get_user_profileSkill未能触发降级,继续阻塞等待,导致整个Agent链路超时。我们紧急修复,增加了redis_client.ping(timeout=200)健康检查,并配置circuit_breaker在连续3次失败后自动熔断。
经验之谈:在Code Review Checklist中,我永远把“是否提供了ContractTest”放在第一条。没有契约测试的Skill,就像没有驾照就上高速——不是能不能开的问题,而是迟早出事的问题。
5. 从零构建一个Production-Ready Skill:以“智能会议纪要生成”为例
理论终需落地。现在,让我们亲手构建一个真实场景下的Skill:generate_meeting_minutes。它接收一段会议录音转写的文本(ASR Result),输出结构化Markdown格式的会议纪要,包含议题摘要、待办事项(Action Items)、责任人(Owner)、截止时间(Due Date)。这个Skill将贯穿前述所有原则,成为可直接投入生产的样板。
5.1 第一步:定义不可妥协的Skill Schema
先写generate_meeting_minutes.skill.yaml,这是整个工程的基石:
name: generate_meeting_minutes version: "1.0.0" description: "将非结构化会议文本提炼为结构化Markdown纪要,自动识别Action Items、Owner、Due Date" input_schema: type: object properties: asr_text: type: string minLength: 100 maxLength: 50000 description: "ASR转写后的原始文本,需包含发言者标记(如'[张三]:今天讨论...'[李四]:我建议...')" meeting_id: type: string pattern: "^MTG_[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" description: "会议唯一标识,用于审计追踪" timezone: type: string enum: ["Asia/Shanghai", "America/New_York", "Europe/London"] default: "Asia/Shanghai" required: ["asr_text", "meeting_id"] output_schema: type: object properties: markdown_content: type: string description: "生成的Markdown格式纪要,必须包含## 议题摘要、## 待办事项两个一级标题" action_items: type: array items: type: object properties: description: type: string description: "待办事项描述,如'整理Q3销售数据'" owner: type: string description: "责任人姓名,需与公司通讯录匹配" due_date: type: string format: date description: "截止日期,格式YYYY-MM-DD" execution_time_ms: type: integer description: "实际执行耗时(毫秒)" required: ["markdown_content", "action_items", "execution_time_ms"]注意几个关键设计点:
asr_text设定了minLength: 100,过滤掉无效的短文本(如“嗯”、“啊”);meeting_id强制UUID格式,确保审计可追溯;action_items数组的每个元素都定义了owner和due_date,杜绝“待办事项无人认领”的模糊输出;execution_time_ms是性能SLO的量化依据,必须由Skill自身精确测量。
5.2 第二步:编写契约驱动的Skill实现
generate_meeting_minutes.py:
from pydantic import BaseModel, ValidationError, validator from typing import List, Dict, Any, Optional import time import re from datetime import datetime, timedelta # 严格遵循output_schema定义的Pydantic Model class ActionItem(BaseModel): description: str owner: str due_date: str @validator('due_date') def validate_due_date(cls, v): try: datetime.strptime(v, "%Y-%m-%d") except ValueError: raise ValueError("due_date must be in YYYY-MM-DD format") return v class SkillOutput(BaseModel): markdown_content: str action_items: List[ActionItem] execution_time_ms: int def generate_meeting_minutes( asr_text: str, meeting_id: str, timezone: str = "Asia/Shanghai" ) -> SkillOutput: start_time = time.time_ns() # Step 1: 输入校验(契约第一道防线) if len(asr_text) < 100: raise ValueError("asr_text too short (<100 chars)") if not re.match(r"^MTG_[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", meeting_id): raise ValueError("invalid meeting_id format") # Step 2: 核心逻辑 - 提炼议题摘要(此处用规则引擎,非LLM,确保确定性) summary = _extract_summary(asr_text) # Step 3: 核心逻辑 - 识别Action Items(正则+NER模型) raw_items = _parse_action_items(asr_text) # 标准化Owner(匹配通讯录) normalized_items = _normalize_owners(raw_items) # 推断Due Date(基于“下周”、“月底”等相对时间) final_items = _infer_due_dates(normalized_items, timezone) # Step 4: 构建Markdown(严格遵循格式要求) markdown = _build_markdown(summary, final_items) # Step 5: 构造输出(Pydantic自动校验) output = SkillOutput( markdown_content=markdown, action_items=final_items, execution_time_ms=int((time.time_ns() - start_time) / 1_000_000) ) return output # 内部函数实现(略,重点在契约与结构) def _extract_summary(text: str) -> str: # 实现细节:提取高频关键词、发言者共识点、结论性语句 pass def _parse_action_items(text: str) -> List[Dict[str, str]]: # 实现细节:匹配"请XXX负责..."、"需要在YYY前完成..."等模式 pass def _normalize_owners(items: List[Dict]) -> List[ActionItem]: # 实现细节:调用HR API,将"小王"、"王工"映射为"王建国" pass def _infer_due_dates(items: List[ActionItem], tz: str) -> List[ActionItem]: # 实现细节:将"下周三"转换为具体日期 pass def _build_markdown(summary: str, items: List[ActionItem]) -> str: md = "## 议题摘要\n\n" + summary + "\n\n## 待办事项\n\n" for i, item in enumerate(items, 1): md += f"{i}. **{item.description}**\n - 责任人:{item.owner}\n - 截止时间:{item.due_date}\n\n" return md关键点解析:
- 输入校验前置:在业务逻辑开始前,用正则和长度检查拦截非法输入,避免无效计算;
- 输出强制Pydantic:
SkillOutput(...)构造时,Pydantic会自动校验due_date格式、action_items数组结构,任何不合规都将抛出ValidationError; - 执行时间精确测量:用
time.time_ns()而非time.time(),避免浮点精度误差; - 所有内部函数命名清晰:
_extract_summary、_parse_action_items,体现单一职责。
5.3 第三步:编写四维测试套件
test_generate_meeting_minutes.py:
import pytest from unittest.mock import patch, MagicMock from generate_meeting_minutes import generate_meeting_minutes, SkillOutput, ActionItem # ContractTest:验证Schema校验 def test_input_schema_validation(): # 测试非法meeting_id with pytest.raises(ValueError, match="invalid meeting_id format"): generate_meeting_minutes( asr_text="test text", meeting_id="INVALID_ID" ) def test_output_schema_validation(): # 测试非法due_date with pytest.raises(ValueError, match="due_date must be in YYYY-MM-DD format"): SkillOutput( markdown_content="# Test", action_items=[ActionItem(description="do it", owner="me", due_date="2024/01/01")], execution_time_ms=100 ) # BoundaryTest:测试边界值 def test_asr_text_min_length(): # 刚好100字符,应通过 valid_text = "A" * 100 result = generate_meeting_minutes(valid_text, "MTG_123e4567-e89b-12d3-a456-426614174000") assert isinstance(result, SkillOutput) def