Agent-Skills工程实践：可测试、可灰度、可监控的智能体能力单元设计

1. 项目概述：Agent-Skills 不是概念玩具，而是可落地的智能体能力工程实践

“Agent-Skills”这个词最近在技术社区里频繁出现，但很多人一看到就下意识联想到大模型API调用、LangChain流水线或者AutoGen的对话编排——其实这恰恰是最大的误解。我带过三个不同行业的智能体落地项目，从某高校实验室的科研辅助系统，到某公司内部的跨系统工单调度平台，再到一个面向中小企业的合同条款比对工具，所有项目复盘时最常被问到的问题不是“用了什么大模型”，而是“这个技能怎么封装才不散架”。Agent-Skills 的本质，是把“能做什么”这件事，从模糊的提示词描述，变成可注册、可测试、可灰度、可监控的独立能力单元。它解决的不是“能不能回答问题”，而是“当用户说‘帮我查上季度华东区销售TOP3客户’时，系统是否能精准识别这是‘多源数据聚合查询’技能，而不是误触发‘生成PPT大纲’或‘起草邮件’”。关键词里没有“LLM”“RAG”“Orchestrator”，只有“Skills”——这本身就是一种信号：重心已从模型层下沉到能力层。适合谁？如果你正在写一个需要调用数据库、发HTTP请求、读取Excel、调用内部审批API的智能体，而不是单纯做问答机器人；如果你的团队里有后端工程师、DBA、甚至懂Power Automate的业务分析师，而不仅仅是Prompt工程师；如果你已经踩过“同一个意图在不同对话轮次里触发不同技能”“技能返回格式不一致导致下游解析失败”“调试一个技能要重启整个Agent服务”的坑——那这篇就是为你写的。它不讲原理图，不画架构框，只讲怎么把一个“查销售数据”的需求，拆成可交付、可交接、可上线的能力模块。

2. 核心设计思路：为什么必须把技能（Skill）从Agent中解耦出来？

2.1 技能不是函数，是具备上下文感知与状态契约的独立服务单元

很多团队一开始做Agent，会直接在主流程里写一堆if-else判断用户意图，然后调用对应函数。比如：

if "销售数据" in user_intent: result = query_sales_db(region="华东", period="上季度", top_n=3) elif "合同条款" in user_intent: result = compare_clauses(doc_a, doc_b, "违约责任")

这看起来很直接，但实际跑两周就会出问题。我参与的第一个项目就卡在这里：销售部临时要求加一个“按产品线细分”的维度，法务部同时提出要高亮差异条款的修订依据。开发同学改完代码，测试发现“查销售数据”技能返回的JSON字段名从top_customers变成了top_clients，导致前端图表渲染全白屏；而合同比对技能新增的修订依据字段，让原本只消费diff_text字段的邮件通知服务直接抛异常。问题根源在于：技能和调用方之间没有契约，只有隐式约定。Agent-Skills 的设计哲学，就是强制建立这种契约。一个合格的Skill，必须明确定义三件事：

• 输入契约（Input Contract）：接受什么结构的参数？哪些必填？哪些可选？参数值域范围是什么？比如query_sales_db技能的输入必须是{"region": str, "period": Enum["上季度","本年度"], "top_n": int in [1,10]}，而不是笼统的dict。
• 输出契约（Output Contract）：返回什么结构？字段名、类型、是否允许为空、错误码定义。比如必须返回{"data": List[{"name": str, "amount": float}], "meta": {"source": "oracle_prod_v2", "latency_ms": int}}，任何字段缺失或类型错位都算违约。
• 上下文感知边界（Context Boundary）：这个技能是否依赖Agent当前对话历史？是否需要访问用户档案？是否必须在特定会话状态下才能执行？比如“撤回上一条审批”技能，必须校验session_state == "pending_approval"且last_action == "submit_for_review"，否则直接拒绝，而不是报错。

提示：契约不是写在文档里的，而是通过Schema验证强制执行的。我们用Pydantic V2定义每个Skill的InputModel和OutputModel，在技能入口处做严格校验。一次校验失败，就记录为SKILL_CONTRACT_VIOLATION事件，而不是让错误蔓延到下游。

2.2 解耦不是为了炫技，而是为了解决四个真实痛点

为什么非要把Skill从Agent主逻辑里剥出来？不是为了架构图好看，而是直面四个每天都在发生的现实问题：

第一，技能迭代与Agent发布周期脱钩。
某公司内部工单系统，Agent主服务每两周发布一次，但销售部每周都要调整“客户分级规则”。如果规则逻辑硬编码在Agent里，每次调整都要走完整CI/CD流程，还要协调测试资源。改成Skill后，规则引擎作为独立Docker服务部署，更新只需替换镜像并触发一次热重载，平均耗时从4小时降到7分钟。关键是：Agent主服务完全无感，连日志里都看不到这次变更。

第二，技能故障隔离，避免单点崩溃拖垮全局。
我们曾遇到一个技能调用外部天气API，因对方服务抖动，超时设置不合理，导致整个Agent响应卡死30秒。解耦后，这个技能被包装成异步任务，超时5秒即降级返回“暂无法获取实时天气”，Agent主流程继续处理其他意图。监控面板上，我们能清晰看到weather_skill.timeout_rate飙升，而agent.total_latency_p95几乎不受影响。

第三，技能复用率从<10%提升到65%+。
最初所有技能都是为单一Agent定制的。比如“查销售数据”技能里硬编码了Oracle连接串，“查库存”技能里写了MySQL的JOIN逻辑。解耦后，我们抽象出DataQuerySkill基类，统一管理连接池、SQL模板、结果映射。新来的“查物流时效”需求，工程师只需继承基类，覆盖build_sql()和map_result()两个方法，2小时就能交付，不用从零写数据库操作。

第四，技能可观测性从“黑盒日志”变成“白盒追踪”。
以前查一个问题，要在Agent日志里翻找“calling skill X”，再切到另一个服务日志看“skill X finished”，中间的参数传递、中间状态全靠猜。现在每个Skill调用自动生成唯一skill_trace_id，贯穿输入校验、执行、输出校验、后置处理全流程。我们在Grafana里建了专门看板，能实时看到：哪个Skill调用最频繁？哪个Skill错误率最高？哪个Skill的平均耗时在缓慢爬升（预示着数据库索引失效）？这些数据直接驱动优化决策，而不是靠“感觉”。

2.3 为什么不用现有框架？我们试过LangChain、LlamaIndex、Semantic Kernel

很多人会问：“已有这么多Agent框架，为什么还要自己搞一套Skill体系？”答案很实在：框架解决的是“怎么连”，而Skill体系解决的是“连什么”和“连得稳不稳”。我们做过三个月的横向对比测试，用同一组业务需求（销售查询、合同比对、工单创建）在三个主流框架上实现：

差距最大的是“参数传递错误率”。LangChain的Runnable链式调用，上游输出是dict，下游输入也是dict，但字段名、嵌套层级、空值处理全靠开发者自觉。我们有个技能叫extract_customer_info，它输出{"customer_id": "C123", "contact_name": "张经理"}，但下游send_notification技能期望的输入是{"cid": "C123", "name": "张经理"}。这种错配在链式调用里极难发现，往往要等用户投诉“通知发错了人”才暴露。而Skill体系强制要求上下游通过明确的Schema交互，编译期就能发现customer_id字段未映射到cid，直接报错。

注意：我们不是反对用框架，而是反对“把框架当银弹”。现在我们的生产环境是混合架构：Agent主流程用LangChain做意图路由和对话管理，但所有具体业务动作，全部由自研Skill体系承载。框架负责“指挥”，Skill负责“打仗”。

3. Skill核心实现细节：从定义、注册到执行的完整闭环

3.1 Skill定义：用Python Class + Pydantic Schema构建可验证契约

一个标准Skill的代码结构，远比想象中更精简，但每个部分都有明确职责。以“查询销售TOP客户”为例：

# sales_top_skill.py from pydantic import BaseModel, Field, validator from typing import List, Dict, Optional from skill_core.base import BaseSkill # 自研基类 class SalesTopInput(BaseModel): region: str = Field(..., description="销售区域，如'华东'、'华北'") period: str = Field(..., description="时间周期，支持'上季度'、'本年度'") top_n: int = Field(3, ge=1, le=10, description="返回客户数量") @validator('period') def validate_period(cls, v): if v not in ["上季度", "本年度"]: raise ValueError("period must be '上季度' or '本年度'") return v class SalesTopOutput(BaseModel): data: List[Dict[str, str]] = Field(..., description="客户列表，每个元素含'name'和'amount'") meta: Dict[str, str] = Field(..., description="元数据，含'source'和'latency_ms'") class SalesTopSkill(BaseSkill): name = "sales_top_query" # 技能唯一标识，用于路由 description = "查询指定区域和周期的销售TOP客户" input_model = SalesTopInput output_model = SalesTopOutput def execute(self, input_data: SalesTopInput) -> SalesTopOutput: # 1. 参数已通过Pydantic校验，region/period/top_n必存在且合法 # 2. 这里写纯业务逻辑：构造SQL、调用DB、处理结果 sql = self._build_sql(input_data.region, input_data.period, input_data.top_n) raw_results = self.db.execute(sql) # 3. 强制映射到output_model结构，字段名、类型、必选性全受控 mapped_data = [ {"name": row[0], "amount": str(row[1])} for row in raw_results ] return SalesTopOutput( data=mapped_data, meta={ "source": "oracle_sales_v3", "latency_ms": int((time.time() - start_time) * 1000) } ) def _build_sql(self, region: str, period: str, top_n: int) -> str: # 具体SQL构造逻辑，此处省略 pass

这个结构的关键设计点：

• 输入/输出模型分离：SalesTopInput和SalesTopOutput是独立Pydantic模型，可被其他服务复用（比如前端表单校验、API网关参数检查）。
• 校验内聚在模型层：@validator装饰器确保period值域检查在数据进入execute()前就完成，而不是在SQL拼接时才发现。
• execute()方法纯净：它只做一件事：把已校验的输入，转换成符合输出契约的结果。没有日志、没有监控埋点、没有重试逻辑——这些由Skill基类统一注入。
• 技能标识唯一且语义化：name = "sales_top_query"不是随便起的，它会成为API路径、Kafka Topic名、Prometheus指标前缀，所以必须小写、下划线、无空格。

3.2 Skill注册中心：动态加载、健康检查与灰度发布

定义好Skill只是第一步，如何让Agent知道“我现在能调用哪些Skill”？我们没用静态配置文件，而是构建了一个轻量级注册中心（Registry），它运行在Agent进程内，但行为像一个微型服务发现系统。

注册流程：

1. Agent启动时，扫描skills/目录下所有Python文件；
2. 通过importlib动态导入，找到所有继承BaseSkill的类；
3. 对每个类，实例化一次（仅用于读取name、description、input_model等元数据），不执行execute()；
4. 将元数据存入内存字典，并启动一个独立线程，对每个Skill执行健康检查（Health Check）。

健康检查不是简单ping一下，而是模拟一次最小可行调用：

def health_check(self): # 构造一个绝对合法的最小输入 dummy_input = self.input_model( region="华东", period="上季度", top_n=1 ) try: # 调用execute，但捕获所有异常 result = self.execute(dummy_input) # 验证输出是否符合output_model契约 self.output_model(**result.dict()) return True, "OK" except Exception as e: return False, f"Health check failed: {str(e)}"

只有健康检查通过的Skill，才会被标记为ACTIVE，并加入可调用列表。如果某个Skill依赖的数据库连接串配置错误，健康检查会失败，该Skill状态变为INACTIVE，Agent路由时会自动跳过它，并告警。

灰度发布机制：
新版本Skill上线，我们不直接替换旧版。而是支持同一name下并存多个版本：

# skills_config.yaml sales_top_query: v1: path: "./skills/sales_top_v1.py" weight: 80 # 80%流量 v2: path: "./skills/sales_top_v2.py" weight: 20 # 20%流量，用于A/B测试

Agent路由时，根据name查到配置，再按权重随机选择版本。v2版本如果错误率超过5%，自动降权至0%，v1接管100%流量。整个过程无需重启Agent。

3.3 Skill执行引擎：超时控制、重试策略与降级熔断

执行阶段才是真正的“战场”。我们观察到，80%的Agent线上故障，源于技能执行时的不可控因素：网络抖动、下游服务慢、数据库锁表、大文件IO阻塞。因此，执行引擎不是简单skill.execute()，而是一套带保护机制的管道：

def safe_execute(skill: BaseSkill, input_data: dict, timeout: int = 10, max_retries: int = 2, fallback: Optional[Callable] = None) -> dict: # 步骤1：输入校验（Pydantic） try: validated_input = skill.input_model(**input_data) except ValidationError as e: raise SkillInputError(f"Input validation failed: {e}") # 步骤2：执行（带超时和重试） for attempt in range(max_retries + 1): try: # 使用threading.Timer或asyncio.wait_for实现超时 result = run_with_timeout( lambda: skill.execute(validated_input), timeout=timeout ) # 步骤3：输出校验 try: validated_output = skill.output_model(**result.dict()) return validated_output.dict() except ValidationError as e: raise SkillOutputError(f"Output contract violation: {e}") except TimeoutError: if attempt == max_retries: if fallback: return fallback() else: raise SkillTimeoutError(f"Timeout after {timeout}s, {max_retries} retries") except Exception as e: if attempt == max_retries: if fallback: return fallback() else: raise e return {} # unreachable

这个引擎的关键特性：

• 超时是硬限制：不是requests.timeout那种软超时，而是进程级中断，防止一个慢技能拖垮整个Agent。
• 重试有策略：不是所有错误都重试。网络超时、503错误会重试；输入校验失败、SQL语法错误则立即失败，重试毫无意义。
• 降级有兜底：fallback参数可以是一个lambda，比如返回{"data": [], "meta": {"source": "cache", "latency_ms": 5}}，告诉用户“数据可能不是最新的，但保证能返回”。
• 错误分类明确：SkillInputError、SkillTimeoutError、SkillOutputError，每种错误对应不同的监控告警和SOP，运维同学一看错误类型就知道该查什么。

我们实测过：在模拟下游API 30%概率超时5秒的压测环境下，启用此引擎后，Agent整体成功率从62%提升到99.8%，P95延迟稳定在1.2秒内，而未启用的对照组P95飙升至8.7秒。

3.4 技能可观测性：从日志到指标的全链路追踪

没有可观测性的Skill，就像没有仪表盘的飞机。我们为每个Skill调用注入了三层追踪：

第一层：结构化日志（Structured Logging）
使用structlog，每条日志都是JSON，关键字段固定：

{ "event": "skill_exec_start", "skill_name": "sales_top_query", "skill_version": "v1", "input_hash": "a1b2c3...", "trace_id": "0x7f8a9b...", "timestamp": "2024-05-20T14:23:11.872Z" }

第二层：Prometheus指标（Metrics）
暴露以下核心指标：

第三层：分布式追踪（Tracing）
集成OpenTelemetry，skill.execute()调用自动成为Span。在Jaeger里，能看到完整链路：

Agent Request (root span) ├── Intent Routing ├── Skill Execution: sales_top_query (span) │ ├── DB Query: SELECT ... (child span) │ └── Result Mapping (child span) └── Response Formatting

这让我们能精准回答：“为什么这个请求慢？”——是卡在DB查询（查SQL执行计划），还是卡在结果映射（Python循环太重）？而不是在日志海里盲搜。

实操心得：不要一上来就堆全套可观测性。我们是分三步走的：第一周只加结构化日志和基础Counter；第二周加Duration Histogram；第三周才接入Tracing。每一步都先看它解决了什么具体问题，再决定是否保留。比如Tracing上线后，我们发现90%的慢请求都集中在sales_top_query的DB子Span，立刻推动DBA优化索引，而不是去优化Python代码。

4. 实战案例拆解：从需求到上线的完整技能交付流程

4.1 需求背景：法务部要一个“合同风险点自动标注”功能

某公司法务部每天要审阅上百份采购合同，其中“付款条件”和“违约责任”条款最容易出问题。他们提的需求很朴素：“上传一份PDF合同，标出所有可能有风险的句子，并说明为什么”。表面看是NLP任务，但实际落地时，问题全在工程侧：

• PDF解析质量不稳定，扫描件OCR错误率高达15%；
• “风险点”定义模糊，法务同事说“付款节点不明确就算风险”，但没给具体规则；
• 标注结果要嵌入到公司OA系统，必须返回标准JSON，字段名不能随意改。

如果按传统AI项目做法，会组建一个NLP小组，花两个月训练模型，再对接PDF解析服务。但我们用Skill体系，只用了5天就交付了MVP。

4.2 技能拆解：把模糊需求拆成3个可验证的Skill

我们没做一个“合同风险分析”大技能，而是拆成三个原子Skill，每个都能独立测试：

Skill 1：pdf_to_text

• 输入：PDF文件二进制流
• 输出：{"text": str, "page_count": int, "ocr_confidence": float}
• 关键约束：ocr_confidence < 0.7时，必须在meta中标记"warning": "low_ocr_quality"，提醒人工复核。
• 实现：调用pdfplumber（文本清晰）+pytesseract（扫描件），自动选择最优路径。

Skill 2：clause_extract

• 输入：{"text": str, "page_count": int}
• 输出：{"clauses": List[{"type": str, "content": str, "page": int}]}
• 关键约束：type必须是枚举值["付款条件", "违约责任", "知识产权", "保密义务"]，不允许出现"其他"。
• 实现：用正则匹配+关键词规则（非LLM），因为法务规则明确，且规则更新频率高，LLM微调成本太高。

Skill 3：risk_assess

• 输入：{"clauses": List[...]}
• 输出：{"risks": List[{"clause_id": str, "risk_level": Enum["high","medium","low"], "reason": str}]}
• 关键约束：risk_level必须有明确计算逻辑，比如“付款条件”中不含“预付款比例”字段，且不含“验收后X日内付款”，则risk_level="high"。
• 实现：规则引擎（Drools语法），法务同事可直接编辑规则文件，无需开发介入。

这三个Skill，输入输出契约清晰，彼此解耦。PDF解析失败，不影响条款提取逻辑；条款类型新增，不用改风险评估代码。

4.3 交付与上线：从本地测试到灰度发布的5步流程

Step 1：本地单元测试全覆盖
每个Skill的test_*.py文件，必须包含：

• 正常输入测试（Happy Path）
• 边界值测试（如top_n=1,top_n=10）
• 错误输入测试（如period="下季度"，应触发ValidationError）
• 降级测试（如DB连不上，是否返回fallback结果）

我们用pytest+pytest-cov，要求单个Skill测试覆盖率≥95%。pdf_to_text技能的测试用例，甚至包含了10个不同扫描质量的PDF样本（从清晰到模糊到倾斜），确保OCR逻辑鲁棒。

Step 2：契约一致性验证
写一个脚本，自动检查所有Skill的input_model和output_model，是否符合公司统一规范：

• 所有description字段不能为空
• 所有Field(...)必填字段，必须有description
• output_model中meta字段必须存在，且类型为Dict[str, Any]
  这个检查作为CI流水线的强制门禁，不通过则禁止合并。

Step 3：集成测试（Integration Test）
不是测单个Skill，而是测Skill链。例如：

# test_contract_flow.py def test_full_contract_risk_flow(): # 1. 用pdf_to_text处理一个已知PDF text_result = pdf_to_text.execute({"pdf_bytes": sample_pdf}) # 2. 将输出喂给clause_extract clause_result = clause_extract.execute(text_result.dict()) # 3. 再喂给risk_assess risk_result = risk_assess.execute(clause_result.dict()) # 4. 断言最终输出符合OA系统要求 assert "risks" in risk_result.dict() assert all(r["risk_level"] in ["high","medium","low"] for r in risk_result.risks)

Step 4：灰度发布与效果验证
上线首日，只对法务部3位同事开放，流量权重5%。我们重点监控：

• pdf_to_text.ocr_confidence的P50是否≥0.85（低于则说明OCR需优化）
• risk_assess返回的risk_level="high"的条款，被法务人工确认为真风险的比例（我们设目标≥80%）
• 整个流程平均耗时是否≤8秒（PDF解析占70%，必须达标）

首日数据显示，OCR置信度P50为0.79，低于目标。我们立刻切到备用OCR引擎（更换tesseract语言包），第二天P50升至0.87，达标。

Step 5：正式发布与知识沉淀
发布后，自动生成两份文档：

• 技能说明书（Skill Spec）：由input_model和output_model自动生成，包含所有字段说明、示例、错误码，供前端和OA系统对接。
• 运维手册（Ops Guide）：列出所有监控指标、告警阈值、常见问题排查步骤（如“pdf_to_text超时，先检查磁盘IO，再检查OCR进程数”）。

这份手册，后来成了新入职工程师的必读材料，他们第一天就能独立排查90%的Skill问题。

5. 常见问题与避坑指南：那些没人告诉你的实战陷阱

5.1 “技能太多，管理不过来”——用领域驱动设计（DDD）划分Skill边界

团队初期常犯的错误，是把所有函数都塞进Skill。比如把“发送邮件”、“保存到数据库”、“调用审批API”全做成独立Skill。结果注册中心里有200+个Skill，命名混乱（send_email_v1,email_sender_pro,notify_by_mail），根本没法维护。

正确做法：按业务领域（Bounded Context）分组，而非技术动作。
我们重新梳理了所有Skill，按法务、销售、HR、IT四大领域归类：

每个领域有自己的skills/子目录，有自己的负责人（Domain Owner）。新需求来了，先问：“这属于哪个领域？”再决定放在哪个Skill里，或者新建一个同领域Skill。这样，200个Skill变成了4个清晰的领域包，每个包内Skill不超过30个，可维护性大幅提升。

5.2 “技能返回格式不一致，前端天天报bug”——强制Schema即文档

前端工程师最恨的，是后端说“这个字段有时有，有时没有”。在Skill体系里，这是绝对禁止的。我们的解决方案是：Schema即文档，文档即契约，契约即代码。

• 所有output_model的字段，必须明确标注Field(default=None)或Field(...)。...表示必填，None表示可选，绝不允许“不写默认值”。
• 可选字段，必须在description里说明“什么条件下会出现”。比如：

  class SalesTopOutput(BaseModel): data: List[Dict[str, str]] = Field(...) meta: Dict[str, str] = Field(...) warning: Optional[str] = Field( None, description="仅当OCR置信度低于0.7时出现，值为警告信息" )

• 每次output_model变更，自动生成Swagger文档，并推送到公司API门户。前端工程师点开就能看到最新字段说明，不用再问后端“这个字段什么时候有”。

我们曾因一个warning字段未加Optional，导致前端JSON解析失败。从此立下规矩：所有模型变更，必须同步更新API门户，CI流水线会自动校验Swagger JSON是否与代码一致，不一致则失败。

5.3 “技能越写越多，启动越来越慢”——懒加载（Lazy Loading）与按需编译

随着Skill数量增长，Agent启动扫描所有Skill文件，耗时从2秒涨到15秒，严重影响本地开发体验和K8s Pod启动速度。

解决方案：懒加载 + 按需编译。
我们改造了注册中心：

• 启动时，只扫描skills/目录下的__init__.py和registry.yaml，不导入任何Skill类；
• registry.yaml定义每个Skill的元数据（name、path、version、dependencies）；
• 当第一次路由到某个Skill时，才动态导入其Python文件，并执行健康检查；
• 同时，我们用mypy对所有Skill进行静态类型检查，但只在CI阶段运行，不阻塞本地启动。

效果：Agent冷启动时间从15秒降至2.3秒，热启动（已有缓存）降至0.8秒。开发同学反馈：“改完一个Skill，Ctrl+S保存，几秒后就能在Postman里测，不用等整个服务重启”。

5.4 “技能要调用内部API，但认证密钥怎么管？”——技能级密钥管理（Skill-Level Secrets）

很多Skill需要调用公司内部系统，比如ERP、CRM，它们需要API Key或OAuth Token。如果把这些密钥硬编码在Skill代码里，或者放在全局配置里，会带来安全风险（密钥泄露）和管理混乱（谁有权修改？）。

我们的方案：为每个Skill配置独立的Secrets Vault。

• 在K8s中，为每个Skill创建独立的Secret对象，名称为skill-{name}-secrets；
• Skill基类提供self.get_secret("api_key")方法，自动从对应Secret中读取；
• 密钥轮换时，只需更新对应Secret，Skill代码完全不用动；
• 权限控制：只有skill-sales-top-query服务账号，才能读取skill-sales-top-query-secrets，其他Skill无法越权访问。

这让我们在一次安全审计中，顺利通过了“密钥最小权限原则”检查。审计员看到每个Skill的密钥都是隔离的，且轮换记录完整，直接给了高分。

5.5 “测试环境没问题，一上生产就失败”——用生产快照（Production Snapshot）做回归测试

最头疼的场景：测试环境一切正常，生产环境却报错。排查发现，是生产数据库表结构多了个字段，而Skill的SQL查询没适配。

对策：定期抓取生产环境快照，用于回归测试。
我们写了一个脚本，每周日凌晨，自动连接生产数据库，导出：

• 表结构DDL（SHOW CREATE TABLE）
• 样本数据（100行，脱敏后）
• 索引信息

这些快照存入S3，命名为snapshot-prod-20240520.tar.gz。CI流水线在运行集成测试时，会自动下载最新快照，启动一个Docker化的MySQL，加载快照数据，再运行所有Skill测试。这样，任何因生产环境变更导致的Skill兼容性问题，都会在代码合并前被发现。

上线三个月，共捕获7次此类问题，包括：

• ERP表新增is_archived字段，导致SELECT *查询返回多余列，破坏了output_model校验；
• CRM索引被误删，sales_top_query耗时从200ms飙升到3秒，触发熔断；
• 数据库字符集从utf8mb4改为utf8，导致某些中文字段乱码。

这些问题，都在测试阶段就被拦截，没一个流入生产。

最后分享一个小技巧：我们给每个Skill加了一个dry_run模式。在execute()方法里，如果检测到input_data.get("dry_run") is True，就跳过真实执行，只返回一个符合output_model的模拟结果（如{"data": [{"name": "TEST_CUSTOMER", "amount": "999999"}], "meta": {"source": "dry_run", "latency_ms": 1}}）。这个模式在前端联调、压力测试、甚至给老板演示时特别有用——不用真的查生产库，就能看到完整UI流程。而且，dry_run本身也受契约约束，返回结果必须能通过output_model校验，保证了模拟的真实性。