企业官网建设流程全解析

1. “agent-skills”不是新词，而是能力建模的底层转向

最近在多个技术社区和内部分享中，频繁看到“agent-skills”这个组合词被单独拎出来讨论——它既不指代某个开源库，也不是某家公司的私有协议，更不是某个SDK的命名空间。我第一次在某跨平台系统的需求评审会上听到这个词时，还以为是拼写错误，结果发现前后五位工程师都在用它描述同一类设计意图：把原本散落在Agent主逻辑里的能力碎片，显式地、可插拔地、带元信息地抽离成独立单元。

这背后其实是一次静默但深刻的范式迁移。过去我们说“这个Agent能查天气、能订会议室、能生成周报”，能力是隐含在代码分支、if-else嵌套或prompt模板里的黑箱行为；而现在，“agent-skills”要求你必须回答三个问题：这个技能的输入契约是什么？它的执行边界在哪里？失败时能否降级或兜底？它不再是一个动词短语（“能做X”），而是一个具备接口定义、状态感知、可观测入口的轻量服务单元。

关键词里虽然空着，但结合当前工程实践，它天然绑定三组核心概念：技能注册中心（Skill Registry）、技能执行上下文（Skill Context）、技能生命周期钩子（Skill Lifecycle Hooks）。这三者共同构成“agent-skills”的事实标准骨架。比如某图像处理Demo中，一个“裁剪图片”技能不能只写个crop_image()函数就完事——它必须声明自己依赖PIL版本≥10.2、接受base64或url两种输入格式、超时阈值设为3s、失败时返回{ "status": "fallback", "reason": "invalid_format" }结构化错误。这些不是附加要求，而是“agent-skills”语义成立的前提。

提示：如果你正在写的Agent代码里还存在“万能函数”（比如一个run_action(action_name, **kwargs)动态分发所有能力），那它本质上还不属于“agent-skills”体系——这只是技能调用的壳，不是技能本身的建模。

这种转向不是为了炫技。我在某高校实验室参与一个教育类Agent项目时深有体会：当业务方提出“希望学生能随时关闭‘解题步骤提示’这个技能，但保留‘答案核对’功能”时，传统写法只能改代码、发版本、等灰度；而采用“agent-skills”架构后，只需在管理后台将step_explanation技能状态切为disabled，5秒内全量生效。能力真正变成了可编排、可开关、可审计的“活单元”。

2. 技能不是函数，是带契约的微型服务

很多人初接触“agent-skills”时，下意识把它等同于“把函数拆得更细”。这是最危险的认知偏差。函数是执行单元，技能是契约单元。两者的根本差异，在于是否定义了可验证的输入输出边界与可预期的非功能约束。

我们以一个真实场景为例：某公司内部知识库Agent需要提供“文档摘要生成”能力。如果按函数思维，可能写出这样的代码：

def generate_summary(doc_id: str, max_length: int = 300) -> str: doc = fetch_doc(doc_id) return llm_summarize(doc.content, max_length)

这段代码在单机调试时完全没问题，但它无法成为“agent-skills”体系中的合法成员。原因有三：

第一，输入契约缺失：doc_id是字符串，但没说明它是否允许为空、是否支持UUID/数字ID/路径格式、长度上限多少。实测中曾因前端传入"doc_abc#123"（含非法字符）导致数据库查询失败，错误堆栈却指向LLM调用层，排查耗时2小时。

第二，非功能约束模糊：max_length=300看似明确，但300个什么？汉字？Token？Unicode码点？不同模型对“长度”定义不同。某次升级HuggingFace模型后，摘要突然变短30%，因为新模型把标点符号也计入token计数，而旧版忽略。

第三，失败处理不可控：llm_summarize()抛出异常时，上层无法区分是网络超时、模型限流还是内容违规。所有错误都被笼统捕获为Exception，日志里只有一行"summary failed"，运维同学根本无法建立告警规则。

真正的“agent-skills”实现，必须显式声明这些契约。以下是该技能在某模拟项目X中的标准定义（YAML格式，实际项目中可转为JSON Schema）：

name: "document_summary" version: "1.2.0" description: "基于文档ID生成指定长度的中文摘要，支持PDF/DOCX/TXT格式" input_schema: type: object required: [doc_id] properties: doc_id: type: string pattern: "^[a-zA-Z0-9_-]{8,32}$" # 严格限制ID格式 description: "文档唯一标识，仅允许字母、数字、下划线、短横线" max_tokens: type: integer minimum: 50 maximum: 500 default: 200 description: "目标摘要token数（按tiktoken cl100k_base计算）" output_schema: type: object required: [summary, token_count] properties: summary: type: string maxLength: 2000 token_count: type: integer description: "实际生成摘要的token数量" source_doc_title: type: string nullable: true lifecycle: timeout_ms: 8000 retry_policy: max_attempts: 2 backoff_factor: 1.5 fallback: enabled: true strategy: "return_empty" condition: "model_rate_limited | network_timeout"

这个定义文件本身就是一个技能的“身份证”。它不包含任何业务逻辑，却决定了该技能能否被注册进技能中心、能否通过自动化校验、能否被前端安全调用。我在某导师指导的课程设计中，曾让A同学和B同学分别实现同一技能，A同学先写代码再补定义，B同学先写定义再写代码。结果B同学的版本上线后零故障，A同学的版本在第三天因doc_id格式变更导致批量失败——因为定义缺失，没人意识到这个字段是强约束。

注意：技能定义文件必须与代码实现分离，且由CI流水线强制校验。我们团队规定，任何未通过skill-validator --strict校验的PR禁止合入。这个工具会检查：输入schema是否覆盖所有函数参数、timeout是否小于全局Agent超时、fallback策略是否匹配错误类型枚举。看似多一步，实则省下90%的线上排查时间。

3. 技能注册中心：让能力从代码走向配置

当“agent-skills”数量超过5个，手工维护调用关系就会失控。某跨平台系统初期只有“查天气”“设提醒”“读邮件”三个技能，开发者直接在Agent主流程里硬编码调用。但当第7个技能“会议纪要生成”加入后，问题开始暴露：新技能需要访问企业微信API，而原有技能共享同一套OAuth token缓存，导致token刷新冲突；第12个技能“合同条款比对”要求GPU加速，但部署节点全是CPU机器——这些本该在设计阶段暴露的问题，全拖到上线后才爆发。

解决方案不是加更多if判断，而是引入技能注册中心（Skill Registry）。它不是一个新服务，而是一套运行时能力治理机制，核心职责只有三件事：统一加载、元数据索引、上下文注入。

3.1 统一加载：从散装import到声明式装配

传统做法中，每个技能都是独立模块，Agent启动时手动import：

from skills.weather import get_weather from skills.reminder import set_reminder from skills.email import read_email # ... 还有10+个类似import

这种方式的问题在于：加载顺序不可控、依赖冲突难发现、热更新不可能。某次紧急修复“邮件解析”技能的正则bug，运维同学重启服务后发现“天气查询”也挂了——因为两个模块都import了不同版本的requests，Python的sys.modules缓存导致高版本覆盖低版本。

“agent-skills”体系要求所有技能通过注册中心统一加载。典型实现是一个SkillLoader类，它扫描指定目录下的技能定义文件（如skills/*.skill.yaml），按依赖关系拓扑排序，再动态导入对应模块：

# skill_loader.py class SkillLoader: def __init__(self, skills_dir: str): self.skills_dir = skills_dir self._registry = {} def load_all(self): # 1. 扫描所有.skill.yaml文件 definitions = self._scan_definitions() # 2. 构建依赖图（如skill_b依赖skill_a的auth模块） dep_graph = self._build_dependency_graph(definitions) # 3. 拓扑排序后依次加载 for skill_def in topological_sort(dep_graph): module = self._import_skill_module(skill_def) self._registry[skill_def.name] = SkillInstance( name=skill_def.name, module=module, definition=skill_def )

关键点在于：技能模块本身不主动注册，而是由注册中心按需实例化。每个技能模块只需暴露一个标准接口：

# skills/weather/__init__.py def execute(context: SkillContext, input_data: dict) -> dict: """所有技能必须实现此函数""" # 实际业务逻辑 pass def validate_input(input_data: dict) -> bool: """可选：输入预校验""" pass

这样，当需要禁用某个技能时，只需在定义文件中将enabled: false，注册中心加载时自动跳过，无需改任何代码。

3.2 元数据索引：让能力可搜索、可审计、可授权

注册中心的核心价值，是把技能从“能用的代码”变成“可管理的资产”。我们给每个技能注入四类元数据：

这些元数据不写死在代码里，而是从.skill.yaml定义文件中提取，并在注册中心启动时构建内存索引。某次生产环境出现CPU飙升，运维同学用curl http://registry:8000/skills?resource=cpu:1000m直接查出占用最高的3个技能，20分钟内完成扩容——这在旧架构中需要登录每台机器ps aux \| grep python逐个排查。

3.3 上下文注入：解决技能间的隐式耦合

技能之间必然存在依赖：比如“发送会议邀请”技能需要先调用“获取用户日历”技能。传统做法是在前者代码里直接import后者，形成硬依赖。这导致两个问题：一是测试困难（必须启动所有依赖技能），二是升级风险（更新日历技能可能意外影响邀请技能）。

“agent-skills”体系通过上下文注入（Context Injection）解耦。注册中心为每个技能创建独立执行上下文，其中预置了其他已注册技能的代理句柄：

class SkillContext: def __init__(self, registry: SkillRegistry): self.registry = registry # 注入所有已注册技能的调用代理 self.skills = { "get_calendar": lambda input_data: registry.invoke("get_calendar", input_data), "send_email": lambda input_data: registry.invoke("send_email", input_data), } def invoke_skill(self, skill_name: str, input_data: dict) -> dict: return self.registry.invoke(skill_name, input_data) # 在技能execute函数中使用 def execute(context: SkillContext, input_data: dict) -> dict: # 无需import，直接通过context调用 calendar_data = context.invoke_skill("get_calendar", {"user_id": input_data["user_id"]}) return send_invitation(calendar_data, input_data)

这个设计的关键在于：技能代码只依赖抽象的SkillContext接口，不依赖具体技能实现。测试时，我们可以轻松Mockcontext.skills["get_calendar"]返回假数据；灰度发布时，可以为send_email技能配置AB测试分流，所有调用它的技能自动受益——因为它们调用的从来不是具体函数，而是注册中心提供的统一入口。

4. 技能执行上下文：隔离、可观测、可重放的生命容器

如果说技能定义是“宪法”，注册中心是“议会”，那么技能执行上下文（Skill Context）就是技能运行的“国家”。它不是简单的参数传递对象，而是承载隔离性、可观测性、可重放性三大特性的运行时容器。很多团队卡在“agent-skills”落地的最后一公里，问题就出在上下文设计过于简陋。

4.1 隔离性：为什么每个技能都需要独立沙箱

我们曾在一个金融类Agent项目中踩过一个经典坑：多个技能共享同一个全局HTTP Session。表面看节省连接复用，实则埋下定时炸弹。“股票行情查询”技能设置session.timeout=5s，“财报PDF下载”技能设置session.timeout=120s。当两个技能并发执行时，后设置的timeout会覆盖前者——导致行情查询请求经常超时失败，错误日志却显示“下游服务响应慢”，误导排查方向。

“agent-skills”要求每个技能拥有独立的执行上下文实例，其核心原则是：上下文内的一切状态，必须是技能专属且不可被其他技能篡改。这包括：

• 网络层隔离：每个上下文持有独立的httpx.AsyncClient实例，配置（timeout、headers、proxies）完全独立；
• 存储层隔离：临时文件写入/tmp/skill_{uuid}/专属目录，避免文件名冲突；
• 状态层隔离：context.state是一个空字典，仅供当前技能暂存中间状态（如“多轮对话中用户确认的选项”），生命周期与本次执行绑定。

实现上，我们采用“上下文工厂模式”：

class SkillContextFactory: @staticmethod def create_for_execution( skill_name: str, execution_id: str, parent_context: Optional[SkillContext] = None ) -> SkillContext: # 1. 创建独立临时目录 temp_dir = f"/tmp/skill_{execution_id}_{int(time.time())}" os.makedirs(temp_dir) # 2. 创建独立HTTP客户端 client = httpx.AsyncClient( timeout=httpx.Timeout( connect=3.0, read=skill_registry.get_timeout(skill_name), # 读超时取技能定义值 write=10.0, pool=5.0 ) ) # 3. 初始化上下文 return SkillContext( execution_id=execution_id, skill_name=skill_name, temp_dir=temp_dir, http_client=client, state={}, # 继承父上下文的只读元数据（如用户ID、会话ID） metadata=parent_context.metadata if parent_context else {} ) # Agent主流程中调用 context = SkillContextFactory.create_for_execution("document_summary", "exec_abc123") result = await skill.execute(context, input_data)

这个设计让技能真正成为“无状态函数”——它不依赖全局变量，不修改共享内存，所有外部依赖都通过上下文注入。某次线上事故中，我们甚至能将一个失败的执行上下文序列化保存，离线重放完全复现问题，因为所有输入、配置、依赖状态都完整封存在这个容器里。

4.2 可观测性：从“黑盒执行”到“白盒追踪”

没有可观测性的技能，就像没有仪表盘的飞机。传统日志只记录"start summary"和"end summary"，中间发生了什么？调用了哪些下游？哪个环节耗时最长？错误是网络问题还是模型拒绝？

“agent-skills”体系要求每个上下文内置结构化追踪能力。我们在SkillContext中集成OpenTelemetry SDK，自动生成符合W3C Trace Context标准的trace_id，并为每个关键操作打点：

class SkillContext: def __init__(self, ...): self.trace_id = generate_trace_id() self.span_stack = [] def start_span(self, name: str, attributes: dict = None): span = { "name": name, "start_time": time.time(), "attributes": attributes or {}, "child_spans": [] } if self.span_stack: self.span_stack[-1]["child_spans"].append(span) self.span_stack.append(span) return span def end_span(self, span: dict, status: str = "OK"): span["end_time"] = time.time() span["duration_ms"] = (span["end_time"] - span["start_time"]) * 1000 span["status"] = status self.span_stack.pop() def to_trace_json(self) -> dict: # 返回符合OTLP格式的trace数据 return {"trace_id": self.trace_id, "spans": self._flatten_spans()}

当技能执行时，代码中插入追踪点：

def execute(context: SkillContext, input_data: dict) -> dict: # 开始主Span main_span = context.start_span("document_summary_main", {"doc_id": input_data["doc_id"]}) try: # 追踪文档获取 fetch_span = context.start_span("fetch_document", {"source": "s3"}) doc = fetch_from_s3(input_data["doc_id"]) context.end_span(fetch_span) # 追踪摘要生成 gen_span = context.start_span("llm_generate", {"model": "gpt-4-turbo"}) summary = call_llm(doc.content, input_data["max_tokens"]) context.end_span(gen_span) context.end_span(main_span) return {"summary": summary} except Exception as e: context.end_span(main_span, status="ERROR") raise e

所有trace数据自动上报到Jaeger或Zipkin。运维同学打开UI，输入trace_id就能看到完整的执行火焰图：清楚看到fetch_document耗时1200ms（S3网络延迟），llm_generate耗时4800ms（模型排队），而技能总耗时6200ms——问题根源一目了然。这比翻1000行日志高效百倍。

4.3 可重放性：为什么“重放”是调试的终极武器

可观测性解决“看到问题”，可重放性解决“复现问题”。在复杂Agent系统中，很多Bug具有强时序依赖：比如“用户连续点击两次提交按钮，第二次请求因token过期失败”。这种场景下，日志里只有"token expired"，但无法知道第一次请求是否成功刷新了token。

“agent-skills”上下文的可重放性，是指将一次执行的完整输入、配置、依赖响应快照保存，后续可100%复现相同执行路径。我们通过三层快照实现：

1. 输入快照：序列化input_data和context.metadata（用户ID、会话ID等）；
2. 配置快照：保存技能定义文件全文 + 当前注册中心配置（超时、重试策略等）；
3. 依赖快照：对所有context.invoke_skill()调用，记录其输入和返回（或错误）。这部分通过代理模式实现：

class SkillInvocationProxy: def __init__(self, registry: SkillRegistry, record_mode: bool = False): self.registry = registry self.record_mode = record_mode self.recorded_calls = [] # 仅在record_mode=True时填充 def invoke(self, skill_name: str, input_data: dict) -> dict: if self.record_mode: # 记录本次调用 recorded_call = { "skill_name": skill_name, "input": copy.deepcopy(input_data), "timestamp": time.time() } try: result = self.registry.invoke(skill_name, input_data) recorded_call["output"] = copy.deepcopy(result) recorded_call["error"] = None except Exception as e: recorded_call["output"] = None recorded_call["error"] = str(e) self.recorded_calls.append(recorded_call) return result else: return self.registry.invoke(skill_name, input_data) # 在上下文中使用 context.invocation_proxy = SkillInvocationProxy(registry, record_mode=True)

当开启录制模式，整个执行过程的所有外部依赖调用都被捕获。保存下来的快照文件（JSON格式）包含：

{ "input": {"doc_id": "doc_xyz789", "max_tokens": 200}, "config": {"timeout_ms": 8000, "retry_policy": {"max_attempts": 2}}, "dependencies": [ { "skill_name": "fetch_document", "input": {"doc_id": "doc_xyz789"}, "output": {"content": "银行年报PDF文本...", "size_bytes": 1245678} }, { "skill_name": "llm_generate", "input": {"text": "银行年报PDF文本...", "max_tokens": 200}, "output": {"summary": "本年度净利润增长12%...", "token_count": 187} } ] }

重放时，SkillInvocationProxy切换为回放模式，所有invoke()调用直接返回快照中的output，完全绕过真实依赖。某次解决“偶发性摘要乱码”问题，我们录制了100次执行，发现只有第37次快照中fetch_document返回的content包含BOM头（\ufeff），从而精准定位到S3上传脚本的编码bug——这种问题在真实环境中几乎无法稳定复现。

5. 技能生命周期钩子：让能力真正“活”起来

技能不是静态函数，而是有生老病死的“活体”。agent-skills体系通过生命周期钩子（Lifecycle Hooks）赋予技能感知自身状态、响应环境变化、优雅退出的能力。很多团队只实现了execute，却忽略了on_load、on_error、on_shutdown这些钩子，导致技能像野草一样疯长，最终难以维护。

5.1 on_load：技能的“出生仪式”

on_load钩子在技能被注册中心加载后、首次执行前触发。它不是初始化函数，而是技能与运行时环境的首次握手。典型用途有三：

第一，资源预热。某些技能依赖冷启动耗时的资源，如大模型tokenizer、向量数据库连接池。若等到execute时再初始化，首请求必然超时。on_load中可提前加载：

def on_load(context: SkillContext): # 预热tokenizer（耗时2s） context.tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese") # 预建连接池（耗时1.5s） context.vector_db = init_vector_pool(max_connections=10) # 标记预热完成 context.set_state("warmup_status", "ready")

第二，权限校验。技能可能需要特定密钥或角色才能启用。on_load中可进行一次性校验：

def on_load(context: SkillContext): api_key = os.getenv("SUMMARY_API_KEY") if not api_key: raise SkillLoadError("Missing SUMMARY_API_KEY environment variable") # 调用密钥验证API response = requests.get("https://api.example.com/validate", headers={"Authorization": f"Bearer {api_key}"}) if response.status_code != 200: raise SkillLoadError(f"API key validation failed: {response.text}")

第三，动态配置加载。技能行为可能依赖外部配置中心（如Consul、Nacos）。on_load中可拉取并缓存：

def on_load(context: SkillContext): config = fetch_config_from_consul("document_summary/v1") context.config = { "max_tokens": config.get("max_tokens", 200), "enable_caching": config.get("enable_caching", True) }

注意：on_load必须是同步、快速执行的。我们团队规定其超时为3秒，超时则技能加载失败。所有耗时操作（如大模型加载）必须异步化，on_load只负责触发异步任务并返回。

5.2 on_error：技能的“急救包”

on_error钩子在execute抛出未捕获异常时触发。它不是try-catch的替代品，而是技能级别的错误应对策略中心。关键在于：它必须返回一个结构化响应，而非继续抛异常。

我们定义了标准错误响应格式：

{ "status": "error", "code": "SKILL_TIMEOUT", "message": "LLM generation timed out after 8000ms", "recovery_suggestion": "Try reducing max_tokens or check model service health", "retryable": true, "fallback_output": null }

on_error的实现逻辑应聚焦三件事：

1. 错误分类：区分SKILL_TIMEOUT（技能自身超时）、DOWNSTREAM_ERROR（依赖服务异常）、INPUT_INVALID（输入校验失败）等；
2. 降级决策：根据错误类型和技能定义中的fallback策略，决定是否启用降级；
3. 可观测增强：记录错误详情到专用指标（如skill_errors_total{skill="document_summary", code="SKILL_TIMEOUT"}）。

典型实现：

def on_error(context: SkillContext, error: Exception, input_data: dict) -> dict: # 1. 分类错误 error_code = classify_error(error) # 2. 检查是否可重试 if error_code in ["NETWORK_TIMEOUT", "DOWNSTREAM_ERROR"] and context.definition.retry_policy.enabled: return { "status": "retrying", "code": error_code, "message": str(error), "retry_after_ms": calculate_backoff(context) } # 3. 启用降级（如果定义中配置了） if context.definition.fallback.enabled and should_fallback(error_code, context.definition.fallback.condition): fallback_result = execute_fallback(context, error_code) return { "status": "fallback", "code": error_code, "message": "Using fallback strategy", "fallback_output": fallback_result } # 4. 返回标准错误 return { "status": "error", "code": error_code, "message": str(error), "recovery_suggestion": get_recovery_suggestion(error_code), "retryable": error_code in ["NETWORK_TIMEOUT", "RATE_LIMITED"] }

这个设计让错误处理变得可预测。前端收到"status": "fallback"就知道可以展示“暂用简化版摘要”，收到"status": "retrying"就自动等待重试，而不是显示“系统错误请重试”这种无效提示。

5.3 on_shutdown：技能的“临终关怀”

on_shutdown钩子在服务进程收到SIGTERM信号、准备优雅退出时触发。它的使命是确保技能释放所有资源，不留下僵尸连接或未完成事务。

常见清理任务：

• 关闭HTTP客户端连接池；
• 刷写本地缓存到持久化存储；
• 向注册中心注销自身（通知其他技能“我即将下线”）；
• 完成正在进行的异步任务（如正在生成的摘要，允许其完成但拒绝新请求）。

实现要点：

def on_shutdown(context: SkillContext): # 1. 标记为不可用（注册中心会停止路由新请求） context.registry.mark_skill_unavailable("document_summary") # 2. 等待进行中的执行完成（最多30秒） wait_for_active_executions(timeout=30) # 3. 关闭资源 if hasattr(context, 'http_client') and context.http_client: asyncio.create_task(context.http_client.aclose()) if hasattr(context, 'vector_db') and context.vector_db: context.vector_db.close() # 4. 刷写缓存 if context.cache: context.cache.flush_to_disk() # 5. 记录优雅退出日志 logger.info("document_summary skill shutdown completed")

某次线上升级，运维同学执行kubectl rollout restart，所有Pod收到SIGTERM。由于每个技能都实现了on_shutdown，整个集群在28秒内完成平滑过渡，零请求失败。而未实现该钩子的旧技能，常因连接池未关闭导致新Pod启动后出现ConnectionRefusedError，需要人工介入。

6. 从零搭建：一个可运行的agent-skills最小可行系统

理论讲完，现在动手搭一个真正能跑起来的最小系统。我们不用任何框架，只用Python标准库和两个轻量依赖（httpx用于HTTP调用，pydantic用于Schema校验），全程代码不超过200行，但已具备“agent-skills”全部核心能力。

6.1 目录结构与依赖

agent-skills-demo/ ├── main.py # Agent主程序 ├── skill_registry.py # 技能注册中心 ├── skill_context.py # 技能上下文实现 ├── skills/ │ └── weather/ │ ├── __init__.py # 技能实现 │ └── weather.skill.yaml # 技能定义 └── pyproject.toml # 依赖声明

pyproject.toml内容：

[build-system] requires = ["setuptools>=45", "wheel"] build-backend = "setuptools.build_meta" [project] name = "agent-skills-demo" version = "0.1.0" dependencies = [ "httpx>=0.24.0", "pydantic>=2.0.0", "pyyaml>=6.0.0" ]

6.2 技能定义文件（weather.skill.yaml）

name: "get_weather" version: "1.0.0" description: "获取指定城市的实时天气" input_schema: type: object required: [city] properties: city: type: string minLength: 2 maxLength: 20 description: "城市名称，如'北京'、'Shanghai'" output_schema: type: object required: [temperature, condition, humidity] properties: temperature: type: number description: "当前温度，摄氏度" condition: type: string enum: ["sunny", "cloudy", "rainy", "snowy"] humidity: type: integer minimum: 0 maximum: 100 lifecycle: timeout_ms: 5000 retry_policy: max_attempts: 2 backoff_factor: 1.5 fallback: enabled: true strategy: "return_stub" condition: "network_timeout | api_unavailable"

6.3 技能实现（skills/weather/init.py）

import json import httpx from typing import Dict, Any from skill_context import SkillContext # 模拟天气API响应 MOCK_WEATHER_DATA = { "北京": {"temperature": 22, "condition": "sunny", "humidity": 45}, "上海": {"temperature": 28, "condition": "cloudy", "humidity": 72}, } def execute(context: SkillContext, input_data: Dict[str, Any]) -> Dict[str, Any]: city = input_data["city"] # 模拟网络调用（实际中用context.http_client） try: # 这里本应调用真实API，为演示用mock if city in MOCK_WEATHER_DATA: return MOCK_WEATHER_DATA[city] else: raise Exception(f"Unknown city: {city}") except Exception as e: # 主动抛出异常，触发on_error raise RuntimeError(f"Weather API call failed: {e}") def on_error(context: SkillContext, error: Exception, input_data: Dict[str, Any]) -> Dict[str, Any]: # 简单降级：返回固定数据 return { "status": "fallback", "code": "API_UNAVAILABLE", "message": "Using fallback weather data", "fallback_output": {"temperature": 25, "condition": "cloudy", "humidity": 60} } def on_load(context: SkillContext): # 模拟加载时校验 if not hasattr(context, 'weather_api_key'): context.weather_api_key = "demo-key-123"

6.4 技能上下文（skill_context.py）

import time import uuid import tempfile import asyncio import httpx from typing import Dict, Any, Optional, Callable from pydantic import BaseModel class SkillContext(BaseModel): execution_id: str skill_name: str temp_dir: str http_client: httpx.AsyncClient state: Dict[str, Any] = {} metadata: Dict[str, Any] = {} class Config: arbitrary_types_allowed = True def __init__(self, **data): super().__init__(**data) # 创建临时目录 self.temp_dir = tempfile.mkdtemp(prefix=f"skill_{self.execution_id}_") def get_state(self, key: str, default=None): return self.state.get(key, default) def set_state(self, key: str, value): self.state[key] = value def invoke_skill(self, skill_name: str, input_data: Dict[str, Any]) -> Dict[str, Any]: # 简化版：实际中调用注册中心 from skill_registry import registry