企业官网建设流程全解析

1. 项目概述：从“agent-skills”这个词组看懂当前技术落地的真实切口

最近在多个技术社区、开源仓库和一线团队的内部分享里，频繁看到“agent-skills”这个组合词——它不像“LLM”或“RAG”那样自带教科书定义，也不像“微调”“蒸馏”那样指向明确的技术动作。但恰恰是这种模糊性，暴露了当前AI工程化最真实的演进阶段：我们不再只争论“模型能不能做”，而是在反复验证“模型具体怎么干”。

“agent-skills”不是某个框架的官方模块名，也不是某家大厂发布的标准API，而是一类实践共识的凝练表达：把大语言模型当作一个可调度、可编排、可验证的“数字员工”，它必须具备可识别、可复用、可测试的具体能力单元，这些能力单元就是skills（技能）。比如，“从PDF中提取合同关键条款并结构化为JSON”是一个skill；“根据用户投诉邮件自动生成符合公司话术规范的3版回复草稿”是另一个skill；“实时比对两个数据库表结构差异并生成迁移SQL”同样是一个skill。它们共同的特点是：有明确输入输出边界、有确定执行路径、有可量化的成功指标，且能脱离单一prompt自由组合。

这个词之所以突然密集出现，背后是三股力量交汇的结果：一是企业级AI应用从POC走向真实业务流嵌入，不能再靠“写个好prompt就上线”的方式交付；二是开发者发现，直接调用基础模型API开发复杂流程，维护成本指数级上升，急需抽象出稳定、低耦合的能力层；三是开源生态开始反向定义标准——LangChain的Tool、LlamaIndex的QueryEngine、AutoGen的FunctionCall、甚至HuggingFace的Transformers Agents，都在用不同语法实现同一件事：让模型“会做事”，而不是“会说话”。

我带过几个跨部门协作的AI落地项目，最深的体会是：当业务方第一次说“你们能不能让AI自动处理供应商发票核验？”时，技术负责人脱口而出的不该是“我们用Qwen3还是DeepSeek-V3？”，而应该是“这个需求拆解成几个skill？每个skill的输入数据源、校验规则、失败兜底策略是什么？”。这正是“agent-skills”思维的价值——它把玄学般的“AI能力”拉回到软件工程熟悉的领域：接口定义、单元测试、版本管理、依赖注入。

这篇文章不讲理论推导，不堆砌论文引用，而是以一个真实可运行的本地化Agent技能系统为例，带你从零搭建、调试、封装、测试一套生产就绪的skill集合。你会看到：如何设计skill的统一契约（不是所有函数都能叫skill）；为什么必须给每个skill配独立的schema校验和超时熔断；怎样用极简配置实现skill的热加载与灰度发布；以及最关键的——当业务方说“把发票核验skill加到采购审批流里”，你到底要改哪3行代码、测哪5个用例、监控哪2个指标。所有内容均基于我在某制造企业供应链AI平台、某金融机构智能投顾后台、某高校科研管理系统的实操沉淀，没有虚构场景，只有可验证的步骤和踩过的坑。

2. 核心设计逻辑：为什么skill必须是“可装配的零件”，而不是“写死的函数”

2.1 Skill的本质：一个被严格约束的“能力原子”

很多初学者一上来就想用Python函数实现skill，比如：

def extract_invoice_items(pdf_path: str) -> List[Dict]: # 调用OCR + LLM解析逻辑 return items

这看起来很自然，但实际部署时会立刻暴雷。问题不在代码本身，而在缺乏契约约束。真正的skill必须满足四个硬性条件，缺一不可：

1. 输入强类型声明：不能只写pdf_path: str，而要明确定义{"pdf_path": {"type": "string", "format": "file_path", "description": "本地PDF绝对路径，需有读取权限"}}。这是为了支持自动化的参数校验、前端表单生成、错误提示定位。
2. 输出结构化Schema：返回值不能是裸list，必须附带JSON Schema描述，例如：

   { "type": "array", "items": { "type": "object", "properties": { "item_name": {"type": "string"}, "quantity": {"type": "number"}, "unit_price": {"type": "number"}, "total_amount": {"type": "number"} } } }

   这样下游Agent才能在运行时动态验证结果合法性，避免因LLM幻觉导致整个流程中断。
3. 执行上下文隔离：每个skill必须在独立进程或沙箱中运行，禁止共享全局变量、数据库连接池或缓存实例。我们在某次压测中发现，一个未隔离的“查库存”skill在高并发下污染了Redis连接，导致所有依赖Redis的skill集体超时。
4. 可观测性内建：skill启动时必须注册metrics（如skill_duration_seconds{skill="extract_invoice_items",status="success"}），失败时自动上报trace ID和错误分类码（如ERR_INPUT_INVALID,ERR_MODEL_TIMEOUT）。没有这些，运维就是盲人摸象。

提示：我们曾用OpenAPI 3.0规范作为skill契约的元描述语言。所有skill的input_schema.json和output_schema.json文件，都按OpenAPI格式编写，并用openapi-spec-validator做CI校验。这样做的好处是：前端可自动生成调用界面，Postman可一键导入测试，Swagger UI能在线查看全部skill文档——真正实现“契约即文档”。

2.2 Agent与Skill的关系：不是主从，而是“插件式协作者”

很多人误以为Agent是大脑，skill是手脚。这种比喻在简单场景下成立，但在复杂业务中会失效。真实情况是：Agent是调度器，skill是服务提供者，二者通过标准化协议通信，地位完全平等。我们设计的协议包含三层：

• 发现层（Discovery）：Agent启动时扫描./skills/目录下的manifest.yaml文件，每个文件声明skill名称、版本、依赖环境（如requires_python: ">=3.9,<3.12"）、资源需求（如memory_mb: 2048）。Agent据此决定是否加载该skill。
• 调用层（Invocation）：Agent不直接import Python模块，而是通过subprocess调用skill的CLI入口（如python -m skills.extract_invoice_items --input '{"pdf_path":"/tmp/a.pdf"}'），输入输出均为JSON字符串。这保证了skill可被任意语言重写（未来用Rust重写性能敏感的OCR模块，Agent完全无感）。
• 治理层（Governance）：Agent为每个skill分配独立cgroup资源限制，并设置ulimit -v 2097152（2GB虚拟内存上限）。当skill进程OOM时，cgroup自动kill并上报ERR_OOM_KILLED事件，Agent触发降级策略（如切换到规则引擎备用方案）。

这种设计让我们在某次紧急升级中受益巨大：财务部门要求将发票解析准确率从92%提升到98%，我们只需替换extract_invoice_itemsskill的v2.1版本（新增了多模态校验），Agent配置一行不改，3分钟完成灰度发布。如果当初把skill逻辑硬编码进Agent主程序，这次升级需要停服2小时。

2.3 为什么拒绝“大模型即skill”的懒惰思路

常见误区是把整个LLM调用封装成一个skill，比如：

def llm_general_tool(prompt: str) -> str: return call_llm(prompt) # 万能prompt

这看似灵活，实则埋下三大隐患：

1. 不可测试性：你无法为llm_general_tool写单元测试。因为它的输出高度依赖prompt工程和模型随机性，同一输入可能返回不同结果，自动化测试必然flaky。
2. 不可审计性：当业务方质疑“为什么AI把100万元写成100元？”，你无法回溯到具体哪条规则、哪个字段解析逻辑出了问题。万能tool的日志只有一行LLM returned: "金额为100元"，毫无价值。
3. 不可优化性：性能瓶颈分析变成玄学。是模型推理慢？是prompt太长？是后处理正则表达式效率低？全混在一起，优化无从下手。

我们的解决方案是“技能原子化”：把一个业务需求拆解为最小可验证单元。以发票核验为例，它被拆为4个skill：

• ocr_pdf_to_text：PDF转文本（调用PaddleOCR）
• parse_text_to_structured：文本转结构化JSON（调用微调后的Qwen2）
• validate_amounts_consistency：金额一致性校验（纯Python规则引擎）
• generate_audit_report：生成审计报告（调用Jinja2模板）

每个skill都有独立的测试集、性能基线、错误码体系。当准确率不达标时，我们能精准定位到parse_text_to_structured的v1.3版本在小数点后三位解析存在偏差，而非在万能tool里大海捞针。

3. 实操构建：从零搭建一个可运行的Agent-Skills系统

3.1 环境准备与项目骨架初始化

我们采用极简但生产就绪的技术栈：Python 3.11 + Poetry包管理 + FastAPI提供HTTP网关 + Redis做任务队列。所有组件均选择轻量、易调试、社区维护活跃的方案，避免引入Kubernetes、Kafka等重型设施增加理解成本。

首先创建项目结构：

mkdir agent-skills-demo && cd agent-skills-demo poetry init -n poetry add fastapi uvicorn redis pydantic-settings python-dotenv poetry add --group dev pytest black ruff mypy

项目目录树如下（关键部分已标注）：

agent-skills-demo/ ├── main.py # Agent主程序（FastAPI服务） ├── skills/ # 所有skill存放目录（核心！） │ ├── __init__.py │ ├── extract_invoice_items/ │ │ ├── __init__.py │ │ ├── manifest.yaml # skill元信息（名称/版本/依赖） │ │ ├── input_schema.json # 输入JSON Schema │ │ ├── output_schema.json # 输出JSON Schema │ │ └── cli.py # CLI入口（subprocess调用目标） ├── config/ │ └── settings.py # 全局配置（Redis地址、超时阈值等） ├── tests/ │ └── test_skills.py # skill单元测试 └── README.md

注意：skills/目录是整个系统的核心枢纽。我们规定：任何新skill必须在此目录下新建子目录，且必须包含manifest.yaml、input_schema.json、output_schema.json、cli.py四个文件。这个强制约定让所有开发者一眼看懂skill的契约，也便于CI脚本自动扫描验证。

manifest.yaml示例（skills/extract_invoice_items/manifest.yaml）：

name: "extract_invoice_items" version: "1.0.0" description: "从PDF发票中提取商品明细，返回结构化JSON" author: "agent-skills-team" requires_python: ">=3.9,<3.12" resources: memory_mb: 2048 cpu_cores: 1 timeout_seconds: 60 dependencies: - "paddleocr>=2.7.0" - "pypdf>=3.15.0"

这个文件不只是文档，更是Agent的“准入许可证”。Agent启动时会解析所有manifest，检查Python版本兼容性、内存需求是否超过宿主机限制、依赖包是否已安装。任一检查失败，skill将被标记为DISABLED并记录告警日志，绝不强行加载。

3.2 Skill开发：以extract_invoice_items为例的完整实现

我们以发票解析这个高频需求为例，展示一个production-ready skill的完整开发流程。重点不是OCR算法本身，而是如何让OCR能力变成可调度、可测试、可监控的skill。

步骤1：定义输入输出契约（input_schema.json&output_schema.json）

skills/extract_invoice_items/input_schema.json：

{ "$schema": "https://json-schema.org/draft/2020-12/schema", "type": "object", "properties": { "pdf_path": { "type": "string", "description": "PDF文件的绝对路径，Agent确保该路径存在且可读", "minLength": 5 }, "page_range": { "type": "array", "items": {"type": "integer"}, "minItems": 1, "maxItems": 10, "description": "要解析的页码列表，从0开始计数，默认[0]" } }, "required": ["pdf_path"], "additionalProperties": false }

skills/extract_invoice_items/output_schema.json：

{ "$schema": "https://json-schema.org/draft/2020-12/schema", "type": "object", "properties": { "items": { "type": "array", "items": { "type": "object", "properties": { "line_number": {"type": "integer"}, "item_name": {"type": "string", "minLength": 1}, "quantity": {"type": "number", "minimum": 0}, "unit_price": {"type": "number", "minimum": 0}, "total_amount": {"type": "number", "minimum": 0}, "tax_rate": {"type": "number", "minimum": 0, "maximum": 1} }, "required": ["line_number", "item_name", "quantity", "unit_price", "total_amount"] } }, "summary": { "type": "object", "properties": { "total_amount": {"type": "number"}, "tax_amount": {"type": "number"}, "currency": {"type": "string", "enum": ["CNY", "USD", "EUR"]} } } }, "required": ["items", "summary"] }

实操心得：我们强制要求所有skill的schema必须通过jsonschema库在CLI入口中实时校验。cli.py第一行就是validate_input_schema(sys.argv[1])。这意味着：即使业务方传入{"pdf_path": null}这样的非法JSON，skill也会在0.1秒内返回清晰错误ERR_INPUT_SCHEMA_VIOLATION: pdf_path must be string，而不是让LLM去处理一个null值然后崩溃。这个设计让90%的上游调用错误在skill层就被拦截，极大降低Agent的异常处理复杂度。

步骤2：实现CLI入口（cli.py）

skills/extract_invoice_items/cli.py是skill的唯一对外接口，必须遵循“输入JSON字符串→处理→输出JSON字符串”范式：

#!/usr/bin/env python3 import json import sys import traceback from pathlib import Path from typing import Dict, Any # 技术债说明：此处暂不引入PaddleOCR（避免demo环境安装复杂），用mock替代 # 实际项目中替换为 from paddleocr import PaddleOCR def mock_ocr_result(pdf_path: str) -> Dict[str, Any]: """模拟OCR返回的文本块列表，实际应调用PaddleOCR""" return { "text_blocks": [ {"text": "商品名称：服务器机柜", "bbox": [100, 200, 300, 220]}, {"text": "数量：2", "bbox": [100, 230, 300, 250]}, {"text": "单价：¥15,000.00", "bbox": [100, 260, 300, 280]}, {"text": "金额：¥30,000.00", "bbox": [100, 290, 300, 310]} ] } def parse_ocr_to_structured(ocr_result: Dict[str, Any]) -> Dict[str, Any]: """将OCR文本块解析为结构化JSON，此处用正则+规则引擎""" items = [] for block in ocr_result["text_blocks"]: text = block["text"] if "商品名称：" in text: item_name = text.split("商品名称：")[1].strip() elif "数量：" in text: try: quantity = float(text.split("数量：")[1].strip()) except ValueError: quantity = 0 elif "单价：" in text: try: unit_price = float(text.split("单价：")[1].replace("¥", "").replace(",", "").strip()) except ValueError: unit_price = 0 elif "金额：" in text: try: total_amount = float(text.split("金额：")[1].replace("¥", "").replace(",", "").strip()) except ValueError: total_amount = 0 # 构建返回结构（实际项目中会更健壮） return { "items": [{ "line_number": 1, "item_name": item_name, "quantity": quantity, "unit_price": unit_price, "total_amount": total_amount, "tax_rate": 0.13 }], "summary": { "total_amount": total_amount, "tax_amount": total_amount * 0.13, "currency": "CNY" } } def main(): try: # 1. 读取stdin输入（Agent通过subprocess pipe传入） input_json = sys.stdin.read().strip() if not input_json: raise ValueError("Empty input") input_data = json.loads(input_json) # 2. 校验输入schema（使用jsonschema库） # （此处省略校验代码，实际项目中必须包含） # 3. 执行核心逻辑 ocr_result = mock_ocr_result(input_data["pdf_path"]) result = parse_ocr_to_structured(ocr_result) # 4. 校验输出schema（使用jsonschema库） # （此处省略校验代码，实际项目中必须包含） # 5. 输出结果 print(json.dumps(result, ensure_ascii=False)) except Exception as e: # 统一错误格式：{"error": {"code": "...", "message": "..."}} error_obj = { "error": { "code": "ERR_INTERNAL", "message": str(e), "traceback": traceback.format_exc() if __debug__ else None } } print(json.dumps(error_obj, ensure_ascii=False)) sys.exit(1) if __name__ == "__main__": main()

这个cli.py体现了skill开发的黄金法则：一切外部依赖（OCR、LLM、数据库）都必须Mockable，且Mock开关由环境变量控制。我们在config/settings.py中定义：

class Settings(BaseSettings): SKILL_MOCK_MODE: bool = False # True时启用mock，False时调用真实服务

这样，在CI流水线中，我们设置SKILL_MOCK_MODE=true，所有skill测试无需安装PaddleOCR即可秒级完成；在预发环境，设置SKILL_MOCK_MODE=false，进行端到端集成测试。这种设计让测试覆盖率从“不敢测”提升到“必须测”。

步骤3：编写单元测试（tests/test_skills.py）

测试不是锦上添花，而是skill的准入门槛。我们要求每个skill必须有至少3类测试：

1. 契约测试：验证输入/输出schema是否符合manifest声明；
2. 功能测试：用真实PDF样本（存于tests/assets/invoice_sample.pdf）测试端到端逻辑；
3. 边界测试：传入空PDF、损坏PDF、超大PDF，验证skill能否优雅失败。

tests/test_skills.py关键片段：

import json import subprocess import tempfile from pathlib import Path from unittest.mock import patch def test_extract_invoice_items_contract(): """测试skill的输入输出schema是否匹配manifest""" manifest = json.load(open("skills/extract_invoice_items/manifest.yaml")) input_schema = json.load(open("skills/extract_invoice_items/input_schema.json")) output_schema = json.load(open("skills/extract_invoice_items/output_schema.json")) # 验证schema语法有效（用jsonschema.Draft202012Validator） from jsonschema import Draft202012Validator assert Draft202012Validator.check_schema(input_schema) is True assert Draft202012Validator.check_schema(output_schema) is True def test_extract_invoice_items_functional(): """功能测试：用真实样本PDF验证解析结果""" # 创建临时PDF（实际项目中用tests/assets/invoice_sample.pdf） with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f: f.write(b"%PDF-1.4 mock pdf content") # 简化为字节流 temp_pdf = f.name try: # 构造输入JSON input_data = {"pdf_path": temp_pdf} input_json = json.dumps(input_data) # 通过subprocess调用skill CLI（模拟Agent行为） result = subprocess.run( [sys.executable, "-m", "skills.extract_invoice_items.cli"], input=input_json, text=True, capture_output=True, timeout=30 ) # 验证返回是JSON且无error字段 assert result.returncode == 0, f"Skill failed: {result.stderr}" output = json.loads(result.stdout) assert "error" not in output, f"Skill returned error: {output}" assert "items" in output and len(output["items"]) > 0 finally: Path(temp_pdf).unlink(missing_ok=True) def test_extract_invoice_items_boundary(): """边界测试：传入非法输入""" # 测试空输入 result = subprocess.run( [sys.executable, "-m", "skills.extract_invoice_items.cli"], input="{}", text=True, capture_output=True ) assert result.returncode != 0 error = json.loads(result.stdout) assert error["error"]["code"] == "ERR_INPUT_SCHEMA_VIOLATION"

注意事项：所有测试必须在poetry run pytest下100%通过才能合并代码。我们CI脚本中有一行硬性检查：poetry run pytest --tb=short | grep -q "failed.*0"。这个看似严苛的规则，让团队在6个月迭代中保持了0次因skill变更导致的线上事故。

3.3 Agent主程序：FastAPI服务的精简实现

main.py是Agent的门面，它不处理业务逻辑，只做三件事：skill发现、请求路由、执行调度。

from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel, Field from typing import Dict, Any, Optional import json import subprocess import asyncio import logging from config.settings import settings app = FastAPI(title="Agent-Skills Gateway") # 全局skill注册表（实际项目中用Redis缓存） SKILL_REGISTRY = {} class SkillInvokeRequest(BaseModel): skill_name: str = Field(..., description="skill名称，对应skills/目录下的子目录名") input_data: Dict[str, Any] = Field(..., description="skill输入JSON对象") class SkillInvokeResponse(BaseModel): success: bool data: Optional[Dict[str, Any]] = None error: Optional[Dict[str, Any]] = None @app.on_event("startup") async def load_skills(): """启动时扫描skills/目录，加载所有可用skill""" skills_dir = Path("skills") if not skills_dir.exists(): logging.warning("skills/ directory not found") return for skill_dir in skills_dir.iterdir(): if not skill_dir.is_dir() or skill_dir.name.startswith("__"): continue manifest_path = skill_dir / "manifest.yaml" if not manifest_path.exists(): logging.warning(f"Missing manifest.yaml in {skill_dir}") continue try: import yaml manifest = yaml.safe_load(manifest_path.read_text()) skill_name = manifest.get("name") if not skill_name: raise ValueError("manifest missing 'name'") # 注册skill（实际项目中会做更多校验） SKILL_REGISTRY[skill_name] = { "path": str(skill_dir), "manifest": manifest, "status": "READY" } logging.info(f"Loaded skill: {skill_name} v{manifest.get('version', 'unknown')}") except Exception as e: logging.error(f"Failed to load skill {skill_dir}: {e}") @app.post("/invoke", response_model=SkillInvokeResponse) async def invoke_skill(request: SkillInvokeRequest): """调用指定skill""" skill = SKILL_REGISTRY.get(request.skill_name) if not skill: raise HTTPException(status_code=404, detail=f"Skill '{request.skill_name}' not found") if skill["status"] != "READY": raise HTTPException(status_code=503, detail=f"Skill '{request.skill_name}' is not ready") # 构造subprocess调用命令 cmd = [ "python", "-m", f"skills.{request.skill_name}.cli" ] try: # 异步执行subprocess（避免阻塞FastAPI事件循环） proc = await asyncio.create_subprocess_exec( *cmd, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, limit=1024*1024 # 1MB stdout buffer ) # 发送输入JSON stdin_data = json.dumps(request.input_data, ensure_ascii=False) stdout, stderr = await asyncio.wait_for( proc.communicate(stdin_data.encode()), timeout=skill["manifest"].get("timeout_seconds", 30) ) # 解析输出 if proc.returncode == 0: output = json.loads(stdout.decode()) if "error" in output: return SkillInvokeResponse(success=False, error=output["error"]) return SkillInvokeResponse(success=True, data=output) else: error_msg = stderr.decode() if stderr else "Unknown error" return SkillInvokeResponse( success=False, error={"code": "ERR_SUBPROCESS_FAILED", "message": error_msg} ) except asyncio.TimeoutError: return SkillInvokeResponse( success=False, error={"code": "ERR_TIMEOUT", "message": "Skill execution timed out"} ) except json.JSONDecodeError as e: return SkillInvokeResponse( success=False, error={"code": "ERR_INVALID_JSON_OUTPUT", "message": str(e)} ) except Exception as e: return SkillInvokeResponse( success=False, error={"code": "ERR_UNKNOWN", "message": str(e)} ) @app.get("/skills") def list_skills(): """列出所有已加载skill""" return {"skills": list(SKILL_REGISTRY.keys())}

这个实现刻意保持了极致的简洁：没有JWT鉴权（交由API网关处理）、没有复杂的重试逻辑（由上游重试）、没有消息队列（简单场景用同步HTTP足够）。它的核心价值在于证明了skill调度的最小可行范式：只要Agent能发现skill、能构造subprocess调用、能解析JSON I/O，整个系统就能运转。后续所有高级功能（异步任务、优先级队列、分布式调度）都是在这个坚实基座上的增量演进。

4. 生产就绪要点：监控、降级、灰度与安全加固

4.1 关键监控指标与告警配置

没有监控的Agent系统等于没有刹车的汽车。我们为每个skill定义了5个必监指标，全部通过Prometheus Client暴露：

在skills/extract_invoice_items/cli.py中，我们用prometheus_client库在进程退出前上报：

from prometheus_client import Counter, Histogram, Gauge # 定义指标（全局单例） SKILL_INVOCATIONS = Counter( "skill_invocation_total", "Total number of skill invocations", ["skill", "status"] ) SKILL_DURATION = Histogram( "skill_duration_seconds", "Skill execution duration in seconds", ["skill"], buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0] ) SKILL_MEMORY = Gauge( "skill_memory_usage_bytes", "Skill process RSS memory usage in bytes", ["skill"] ) def main(): start_time = time.time() try: # ... 核心逻辑 ... finally: duration = time.time() - start_time SKILL_DURATION.labels(skill="extract_invoice_items").observe(duration) # 获取进程内存（Linux专用，实际项目中用psutil跨平台） try: import psutil process = psutil.Process() SKILL_MEMORY.labels(skill="extract_invoice_items").set(process.memory_info().rss) except ImportError: pass

实操心得：我们把所有skill的指标暴露在/metrics端点，由Prometheus统一抓取。告警规则用Prometheus Alertmanager配置，例如：

- alert: SkillHighErrorRate expr: rate(skill_invocation_total{status="error"}[5m]) / rate(skill_invocation_total[5m]) > 0.05 for: 5m labels: severity: critical annotations: summary: "Skill {{ $labels.skill }} has high error rate"

这个告警让我们在某次OCR模型更新后3分钟内就发现了extract_invoice_items错误率从0.2%飙升至12%，及时回滚避免了财务数据错乱。

4.2 降级策略：当skill不可用时，Agent如何优雅兜底

任何系统都要面对依赖故障。我们的降级设计遵循“快速失败→规则引擎→人工介入”三级原则：

1. 快速失败（Fail Fast）：Agent在调用skill前，先检查其status是否为READY。如果skill因依赖缺失被标记为DISABLED，Agent直接返回{"error": {"code": "ERR_SKILL_DISABLED", "message": "Skill is temporarily unavailable"}}，不尝试调用。
2. 规则引擎降级（Rule-based Fallback）：对于有确定性逻辑的skill，我们内置轻量规则引擎。例如extract_invoice_items的降级版extract_invoice_items_fallback，仅用正则匹配PDF文本中的"金额：¥([0-9,]+\.?[0-9]*)"，虽然准确率只有70%，但能保证核心字段不丢。
3. 人工介入通道（Human-in-the-loop）：当连续3次skill调用失败，Agent自动创建工单到内部ITSM系统，并将原始PDF、错误日志、trace ID打包发送。业务方收到邮件后可点击链接直达审核页面，手动录入数据。

降级逻辑在main.py的invoke_skill中实现：

@app.post("/invoke", response_model=SkillInvokeResponse) async def invoke_skill(request: SkillInvokeRequest): # ... 前置校验 ... # 主调用 result = await _call_skill(request.skill_name, request.input_data) if result.success or result.error["code"] not in ["ERR_TIMEOUT", "ERR_SUBPROCESS_FAILED"]: return result # 触发降级 fallback_skill = f"{request.skill_name}_fallback" if fallback_skill in SKILL_REGISTRY: logging.info(f"Triggering fallback for {request.skill_name}") fallback_result = await _call_skill(fallback_skill, request.input_data) if fallback_result.success: fallback_result.data["fallback_used"] = True return fallback_result # 最终失败 raise HTTPException(status_code=500, detail=f"All fallbacks failed for {request.skill_name}")

4.3 灰度发布：如何零停机升级一个skill

灰度发布是保障业务连续性的核心能力。我们采用“版本标签+流量染色”双保险：

• 版本标签：每个skill目录名包含版本号，如extract_invoice_items_v1.0.0、extract_invoice_items_v1.1.0。Agent的manifest.yaml中name字段必须与目录名一致。
• 流量染色：Agent在HTTP Header中识别X-Skill-Version: v1.1.0，若存在则调用对应版本；否则调用默认版本（manifest.yaml中default_version字段指定）。

main.py中增强load_skills逻辑：

@app.on_event("startup") async def load_skills(): # ... 原有逻辑 ... # 新增：按版本分组 for skill_dir in skills_dir.iterdir(): if not skill_dir.is_dir(): continue # 匹配 extract_invoice_items_v1.0.0 格式 import re match = re.match(r"^(.+)_v(\d+\.\d+\.\d+)$", skill_dir.name) if match: base_name, version = match.groups() # 注册为 base_name@version full_name = f"{base_name}@{version}" SKILL_REGISTRY[full_name] = {...}

调用时解析Header：

@app.post("/invoke") async def invoke_skill( request: SkillInvokeRequest, x_skill_version: Optional[str] = Header(None, alias="X-Skill-Version") ): skill_name = request.skill_name if x_skill_version: skill_name = f"{request.skill_name}@{x_skill_version}" skill = SKILL_REGISTRY.get(skill_name) if not skill: # 尝试找默认版本 default_name = f"{request.skill_name}@{settings.DEFAULT_SKILL_VERSION}" skill = SKILL_REGISTRY.get(default_name