企业官网建设流程全解析

1. 项目概述：这不是一个“技能库”，而是一套可插拔的智能体行为引擎

“agent-skills”这个名称乍看像某个开源项目的子模块，或是某篇论文里被轻描淡写带过的术语。但在我过去三年深度参与多个智能体（Agent）系统落地项目的过程中，这个词反复出现在架构评审会、调试日志和凌晨三点的 Slack 消息里——它从来不是附庸，而是决定一个 Agent 能不能真正“做事”的分水岭。核心关键词就藏在这四个字母里：agent（智能体）、skills（技能）。它指的不是大模型本身的知识储备，而是让模型能调用外部工具、感知真实环境、执行具体动作的一整套可注册、可编排、可验证的行为能力单元。比如，当用户说“把上周销售数据导出成 Excel 并发给张经理”，真正的难点不在于理解这句话，而在于：调哪个 API 获取数据？用什么库生成 Excel？邮箱地址从哪查？附件大小超限怎么处理？这些就是 skills 要解决的问题。它适合三类人：正在从 Prompt 工程转向真实 Agent 开发的工程师；需要评估 Agent 产品落地边界的业务方；以及想搞懂“为什么我的 ChatGPT 插件总在关键步骤卡住”的技术决策者。它不教你怎么写 prompt，而是告诉你，当 prompt 失效时，该往哪个方向补上真实的能力。

我第一次意识到 skills 的重要性，是在某高校实验室做的一个跨平台文档协同 Demo。当时模型能完美总结 PDF 内容，但一到“把这份总结同步到指定 Notion 数据库并打上‘已审阅’标签”这一步就彻底失联。我们花了两天排查模型输出格式，最后发现根本问题在于：没有定义一个notion_update_pageskill，更没做字段映射校验和权限 token 的自动轮换。后来我们把所有操作抽象成 7 个原子 skill：fetch_pdf_content、extract_text、call_llm_for_summary、search_notion_db、create_notion_page、update_notion_page、send_email_with_attachment。每个 skill 都有独立的输入 schema、错误重试策略、超时控制和日志埋点。结果是，整个流程的失败率从 68% 降到 4.2%，而且每次报错都能准确定位到是search_notion_db的 API 返回了空数组，而不是笼统的“LLM 调用失败”。这说明，“agent-skills”本质是一套面向生产环境的行为契约体系——它用结构化接口约束了模型的“幻想空间”，把不可控的自然语言输出，锚定在可控的程序执行路径上。

2. 整体设计与思路拆解：为什么必须放弃“万能函数”，拥抱“技能插槽”

很多人初学 Agent 开发时，会本能地写一个巨无霸函数，比如execute_action(action_name, **kwargs)，然后在里面用 if-elif 堆砌所有可能的操作。我试过，也踩过坑。在某公司内部知识库项目中，我们最初用这种方式集成了 12 个功能：查文档、改权限、发通知、调 BI 接口、生成图表……代码行数不到 500，但上线一周后，运维告警里全是KeyError: 'chart_type'和TimeoutError: waiting for BI response。问题不在逻辑，而在设计哲学：这种“大杂烩”模式把协议层、执行层、错误处理层全搅在一起，导致任何一个微小变更（比如 BI 接口加了个新参数）都得动全局函数，测试成本指数级上升。后来我们彻底重构，核心就一条：每个 skill 必须是一个独立、自洽、可验证的最小执行单元。这直接决定了整个系统的可维护性和扩展性。

为什么非得这么麻烦？因为真实世界里的操作天然具备三个刚性特征：异构性、状态依赖、失败常态。异构性指不同系统 API 差异巨大——Notion 用 OAuth2，内部 CRM 用 JWT，老系统甚至要模拟表单 POST；状态依赖指操作有严格顺序，比如“删服务器”前必须“备份数据”，而“备份”又依赖“确认磁盘空间足够”；失败常态则是现实——网络抖动、限流、下游服务降级，都是家常便饭。一个 skill 如果不显式声明自己的前置条件（precondition）、副作用（side effects）、重试策略（retry policy）和降级方案（fallback），那它在复杂流程里就是一颗定时炸弹。我们最终采用的方案是“三层契约模型”：第一层是Schema 层，用 JSON Schema 定义 skill 的输入/输出结构，强制类型检查和必填项校验；第二层是执行层，每个 skill 封装一个独立函数，只做一件事，且必须返回标准 Result 对象（含 success、data、error、log）；第三层是治理层，由统一的 Skill Registry 管理所有 skill 的注册、启停、版本灰度和调用链追踪。这个设计不是为了炫技，而是为了解决一个实际问题：当业务方突然要求“给所有导出 Excel 的操作加水印”，你能在 5 分钟内定位到generate_excel这个 skill，修改其内部实现，而不影响send_email或fetch_data的任何一行代码。这才是工程化的起点。

提示：不要试图用装饰器或元编程“自动注册”所有函数。我们早期用@register_skill装饰器，结果发现调试时 IDE 找不到函数引用，单元测试 mock 困难，CI 流程里还因 import 顺序问题偶发注册失败。现在坚持手动注册：在skills/__init__.py里显式调用registry.register("send_email", send_email_skill)。看似多写两行，但所有 skill 的入口、版本、作者都在一个文件里可查，新人上手成本直降 70%。

3. 核心细节解析与实操要点：从定义到验证，一个 skill 的完整生命周期

定义一个真正可用的 skill，远不止写个函数那么简单。以最常用的web_searchskill 为例，它的表面功能是“搜索网页”，但实际要覆盖的细节远超想象。我们先看一个典型失败案例：某团队直接封装了 SerpAPI 的 Python SDK，函数签名是def web_search(query: str) -> List[str]。上线后用户反馈“搜不到最新新闻”，排查发现是 SerpAPI 默认只返回前 10 条，且缓存时间长达 24 小时。这暴露了 skill 设计的致命缺陷：没有暴露关键控制参数，也没有声明能力边界。一个合格的web_searchskill 至少要包含以下要素：

3.1 输入 Schema 的严谨设计

我们用 Pydantic V2 定义输入模型：

from pydantic import BaseModel, Field from typing import Optional, List class WebSearchInput(BaseModel): query: str = Field(..., description="搜索关键词，支持引号精确匹配") num_results: int = Field(5, ge=1, le=50, description="返回结果数量，1-50") time_range: Optional[str] = Field( None, description="时间范围，如 'q1_2024'（2024年第一季度）, 'past_week', 'past_month'" ) site_filter: Optional[str] = Field( None, description="限定域名，如 'github.com'，支持通配符 '*.org'" ) safe_search: bool = Field(True, description="是否启用安全搜索过滤成人内容")

注意这里每个字段都有明确的业务含义和约束（ge=1, le=50），而非简单int。time_range的枚举值不是硬编码，而是从配置中心动态加载，方便后续对接不同搜索引擎。这种设计让 LLM 在规划（planning）阶段就能准确生成符合要求的参数，避免因参数越界导致 skill 直接崩溃。

3.2 执行逻辑的防御性编程

真正的难点在函数体内。我们不会直接调用serpapi.search()，而是构建一个带熔断和降级的执行链：

def web_search(input: WebSearchInput) -> SkillResult: # 步骤1：前置校验（防止无效请求） if len(input.query.strip()) < 2: return SkillResult.failure("query too short, minimum 2 chars") # 步骤2：构建请求参数（适配多引擎） params = { "q": input.query, "num": input.num_results, "safe": "active" if input.safe_search else "off" } if input.time_range: params["tbs"] = time_range_to_tbs(input.time_range) # 转换为 Google tbs 格式 # 步骤3：熔断器调用（基于 Redis 记录失败率） try: with circuit_breaker("serpapi", failure_threshold=3, timeout=60): response = serpapi_client.search(params) except CircuitBreakerOpenError: # 降级：用本地缓存 + 关键词匹配（精度低但可用） cached = cache.get(f"search:{input.query}") if cached: return SkillResult.success(cached[:input.num_results]) return SkillResult.failure("service unavailable, fallback failed") # 步骤4：结果清洗（统一输出结构） results = [] for item in response.get("organic_results", [])[:input.num_results]: results.append({ "title": item.get("title", ""), "url": item.get("link", ""), "snippet": item.get("snippet", "")[:200] # 截断防爆内存 }) return SkillResult.success(results)

这里的关键是：每个 skill 都要自己处理超时、熔断、降级、数据清洗。我们曾遇到过 SerpAPI 返回 500 但 SDK 未抛异常的情况，导致 skill 卡死。后来在serpapi_client封装层加了强制timeout=30和raise_for_status()，才彻底解决。这再次证明：skill 不是胶水代码，而是有独立生命体征的“数字工人”。

3.3 输出验证与可观测性埋点

Skill 的输出必须可验证。我们要求所有 skill 的SkillResult.success(data)中，data必须是 JSON-serializable 的 dict/list，且通过预设的输出 Schema 校验：

class WebSearchOutput(BaseModel): results: List[Dict[str, str]] = Field(..., min_items=0, max_items=50) # 每个 result 必须有 title/url/snippet @validator('results') def validate_result_fields(cls, v): for r in v: if not all(k in r for k in ["title", "url", "snippet"]): raise ValueError("each result must have title, url, snippet") return v

同时，每个 skill 调用都会记录结构化日志：

logger.info("skill_executed", skill_name="web_search", input_hash=hashlib.md5(str(input.dict()).encode()).hexdigest(), duration_ms=elapsed_ms, result_size=len(results), status="success" if success else "failed" )

这些日志被接入 ELK，运营同学能直接查“过去 24 小时web_search的平均耗时和失败率”，而不用翻代码。这才是 production-ready 的 skill。

4. 实操过程与核心环节实现：从零搭建一个可运行的 skills 系统

现在我们动手搭建一个最小可行系统（MVP），目标：让一个 LLM 能通过调用 skills 完成“查询北京今日天气并发送短信提醒”。整个过程分为四步：环境准备、skill 编写、Agent 集成、流程编排。所有代码基于 Python 3.11+，不依赖任何商业框架，确保你能完全掌控每个环节。

4.1 环境准备：轻量但完备的依赖栈

我们刻意避开 LangChain 等大框架，选择更底层的组合：

• LLM 接口：openai>=1.0.0（兼容 OpenAI 兼容 API，如 Ollama、vLLM）
• 技能注册与调度：pydantic>=2.0.0（Schema 验证）、tenacity>=8.0.0（重试）、cachetools>=5.0.0（本地缓存）
• HTTP 客户端：httpx>=0.24.0（异步友好，支持 HTTP/2）
• 配置管理：pydantic-settings>=2.0.0（环境变量优先的配置）

创建pyproject.toml：

[build-system] requires = ["hatchling"] build-backend = "hatchling.build" [project] name = "agent-skills-mvp" version = "0.1.0" dependencies = [ "openai>=1.0.0", "pydantic>=2.0.0", "tenacity>=8.0.0", "cachetools>=5.0.0", "httpx>=0.24.0", "pydantic-settings>=2.0.0", ] [project.optional-dependencies] dev = ["pytest>=7.0.0", "black>=23.0.0"]

安装后，初始化项目结构：

agent-skills-mvp/ ├── main.py # Agent 主流程 ├── skills/ │ ├── __init__.py # 技能注册入口 │ ├── weather.py # 天气查询 skill │ └── sms.py # 短信发送 skill ├── config/ │ └── settings.py # 配置管理 └── utils/ └── registry.py # 技能注册中心

4.2 编写两个核心 skill：weather 和 sms

先看skills/weather.py。我们用和风天气免费 API（需申请 key）：

import httpx from pydantic import BaseModel, Field from tenacity import retry, stop_after_attempt, wait_exponential from typing import Optional, Dict, Any class WeatherInput(BaseModel): city: str = Field(..., description="城市名称，如 '北京'") unit: str = Field("c", description="温度单位，'c' 或 'f'") class WeatherOutput(BaseModel): city: str temperature: float condition: str humidity: int wind_speed: float # 使用 httpx.AsyncClient 复用连接池 weather_client = httpx.AsyncClient( base_url="https://devapi.qweather.com/v7/weather/", timeout=httpx.Timeout(10.0, connect=5.0) ) @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), reraise=True ) async def get_weather(input: WeatherInput) -> Dict[str, Any]: # 步骤1：通过城市名获取 location_id（和风要求） loc_resp = await httpx.get( "https://geoapi.qweather.com/v2/city/lookup", params={"key": "YOUR_KEY", "location": input.city} ) loc_resp.raise_for_status() loc_data = loc_resp.json() if not loc_data.get("location"): raise ValueError(f"city {input.city} not found") location_id = loc_data["location"][0]["id"] # 步骤2：获取实时天气 weather_resp = await httpx.get( f"now", params={"key": "YOUR_KEY", "location": location_id} ) weather_resp.raise_for_status() data = weather_resp.json()["now"] return { "city": input.city, "temperature": float(data["temp"]), "condition": data["textDay"], "humidity": int(data["humidity"]), "wind_speed": float(data["windSpeed"]) } # Skill 包装函数（符合统一接口） async def weather_skill(input: WeatherInput) -> SkillResult: try: raw_data = await get_weather(input) # 输出验证 output = WeatherOutput(**raw_data) return SkillResult.success(output.model_dump()) except Exception as e: return SkillResult.failure(f"weather api error: {str(e)}")

注意：@retry装饰器放在get_weather上而非 skill 函数上，因为重试应针对 HTTP 请求，而非整个 skill 逻辑（比如输入校验失败不该重试）。weather_client是全局复用的，避免频繁创建连接。

再看skills/sms.py，用阿里云短信服务（需开通）：

from aliyunsdkcore.client import AcsClient from aliyunsdkcore.auth.credentials import AccessKeyCredential from aliyunsdkdysmsapi.request.v20170525 import SendSmsRequest import json class SmsInput(BaseModel): phone: str = Field(..., description="手机号，11位数字") content: str = Field(..., description="短信内容，不超过 500 字") # 阿里云客户端（单例） sms_client = AcsClient( ak="YOUR_AK", secret="YOUR_SECRET", region_id="cn-hangzhou" ) def sms_skill(input: SmsInput) -> SkillResult: try: # 校验手机号 if not input.phone.isdigit() or len(input.phone) != 11: return SkillResult.failure("invalid phone number format") # 构建短信请求（阿里云要求模板） request = SendSmsRequest.SendSmsRequest() request.set_accept_format('json') request.set_SignName("测试签名") # 替换为你的签名 request.set_TemplateCode("SMS_123456") # 替换为你的模板ID request.set_PhoneNumbers(input.phone) # 阿里云模板变量格式：{"code":"1234"}，我们简单处理 request.set_TemplateParam(json.dumps({"content": input.content})) response = sms_client.do_action_with_exception(request) resp_json = json.loads(response) if resp_json.get("Code") == "OK": return SkillResult.success({"message_id": resp_json.get("MessageId")}) else: return SkillResult.failure(f"sms send failed: {resp_json.get('Message')}") except Exception as e: return SkillResult.failure(f"sms client error: {str(e)}")

这里的关键是：skill 必须处理领域特定的协议细节。阿里云要求模板 ID、签名、JSON 参数格式，这些都不能丢给 LLM 去猜，必须在 skill 内部封装好。

4.3 Agent 集成：让 LLM 学会“调用技能”

核心是main.py中的 Agent 循环。我们不用复杂推理框架，用最朴素的“Thought-Action-Observation”循环：

import asyncio from openai import AsyncOpenAI from skills import registry # 导入注册中心 from utils.registry import SkillRegistry # 初始化技能注册中心 skill_registry = SkillRegistry() # 手动注册技能（见 2.2 节提示） skill_registry.register("get_weather", weather_skill) skill_registry.register("send_sms", sms_skill) # 初始化 LLM 客户端 client = AsyncOpenAI( base_url="http://localhost:11434/v1", # Ollama 本地服务 api_key="ollama" ) async def run_agent(user_query: str): # Step 1: LLM 规划（生成 tool call） plan_prompt = f""" 你是一个智能助手，能调用以下技能： - get_weather(city: str, unit: str): 查询指定城市天气 - send_sms(phone: str, content: str): 发送短信 用户请求：{user_query} 请按 JSON 格式输出你要调用的技能，例如： {{"tool": "get_weather", "params": {{"city": "北京", "unit": "c"}}}} 或多个技能（按顺序）： [{{"tool": "get_weather", ...}}, {{"tool": "send_sms", ...}}] 只输出 JSON，不要任何解释。 """ plan_response = await client.chat.completions.create( model="qwen2:7b", # 本地模型 messages=[{"role": "user", "content": plan_prompt}], temperature=0.0 ) plan_json = plan_response.choices[0].message.content # Step 2: 解析并执行技能 try: plan = json.loads(plan_json) if isinstance(plan, dict): plan = [plan] # 统一为列表 observations = [] for action in plan: tool_name = action.get("tool") params = action.get("params", {}) if not tool_name or tool_name not in skill_registry.skills: observations.append(f"ERROR: unknown tool '{tool_name}'") continue # 调用 skill skill_func = skill_registry.skills[tool_name] try: # 同步 skill 用 run_sync，异步用 await if asyncio.iscoroutinefunction(skill_func): result = await skill_func(**params) else: result = skill_func(**params) obs = f"SUCCESS: {tool_name}({params}) -> {result.data}" observations.append(obs) except Exception as e: observations.append(f"ERROR: {tool_name}({params}) -> {str(e)}") # Step 3: LLM 总结（将 observations 转为自然语言回复） summary_prompt = f""" 用户请求：{user_query} 技能执行结果： {'\n'.join(observations)} 请用简洁中文总结结果，不要提技术细节。如果全部成功，直接给出答案；如果有失败，说明原因。 """ summary_response = await client.chat.completions.create( model="qwen2:7b", messages=[{"role": "user", "content": summary_prompt}], temperature=0.3 ) return summary_response.choices[0].message.content except json.JSONDecodeError: return "无法理解请求，请重试。" except Exception as e: return f"系统错误：{str(e)}" # 运行示例 if __name__ == "__main__": result = asyncio.run(run_agent("查询北京今日天气，并发送短信提醒我")) print(result)

这个 Agent 的精妙之处在于：它把“技能调用”变成了 LLM 的一个标准输出格式任务，而不是靠复杂插件机制。我们用专门的 prompt 引导 LLM 输出 JSON，再由 Python 解析执行。这样做的好处是：调试极其简单——你可以把plan_prompt单独喂给 LLM，看它是否能正确生成{"tool": "get_weather", "params": {"city": "北京"}}。如果不行，问题就在 prompt 或模型能力，而不是框架 bug。

4.4 流程编排：超越线性调用的技能协作

真实场景中，skills 往往需要协作。比如“订会议室”需要：查日历空闲 → 创建事件 → 发邀请邮件 → 同步到 IM。我们设计了一个轻量编排器WorkflowEngine：

from typing import List, Dict, Any, Callable class WorkflowStep: def __init__(self, skill_name: str, params: Dict[str, Any], next_on_success: str = None, next_on_failure: str = None): self.skill_name = skill_name self.params = params self.next_on_success = next_on_success self.next_on_failure = next_on_failure class WorkflowEngine: def __init__(self, registry: SkillRegistry): self.registry = registry async def execute(self, steps: List[WorkflowStep]) -> Dict[str, Any]: context = {} # 共享上下文，用于参数传递 for step in steps: # 动态解析参数中的上下文变量，如 "{weather.temperature}" resolved_params = self._resolve_context(step.params, context) try: skill_func = self.registry.skills[step.skill_name] if asyncio.iscoroutinefunction(skill_func): result = await skill_func(**resolved_params) else: result = skill_func(**resolved_params) # 将结果存入上下文，供后续步骤使用 context[f"{step.skill_name}_result"] = result.data # 根据结果跳转 if result.success and step.next_on_success: continue # 下一步 elif not result.success and step.next_on_failure: # 跳转到失败分支 pass except Exception as e: context[f"{step.skill_name}_error"] = str(e) break return context # 使用示例：天气+短信工作流 workflow = WorkflowEngine(skill_registry) steps = [ WorkflowStep( skill_name="get_weather", params={"city": "北京"}, next_on_success="send_sms" ), WorkflowStep( skill_name="send_sms", params={"phone": "13800138000", "content": "北京今日天气：{get_weather_result.temperature}°C，{get_weather_result.condition}"} ) ] result = await workflow.execute(steps)

这里"{get_weather_result.temperature}"的语法糖，是我们在_resolve_context中实现的 Jinja2 风格模板渲染。它让 skills 能自然串联，而无需 LLM 去记忆中间结果。这才是“agent-skills”作为行为引擎的真正威力——它把复杂的多步协作，降维成清晰的、可配置的、可审计的流程图。

5. 常见问题与排查技巧实录：那些只有踩过才知道的坑

在十几个项目中，我们整理出一份高频问题清单，每一条都来自血泪教训。这些问题往往不会出现在官方文档里，但会实实在在卡住你的进度。

5.1 LLM 总是生成不存在的 skill 名称

现象：Agent 日志显示ERROR: unknown tool 'get_weater'（拼错）或send_email_v2（版本号乱加）。
根因：LLM 的输出受训练数据影响，对 skill 名称的“拼写稳定性”极差。我们曾统计过，同一 prompt 下，Qwen2-7B 有 12% 的概率把get_weather写成get_weatcher。
解决方案：

• 建立 skill 别名映射表：在注册时，为每个 skill 添加常用别名：

  skill_registry.register("get_weather", weather_skill, aliases=["weather", "forecast", "check_weather"])

• 在解析前做模糊匹配：用difflib.get_close_matches(tool_name, list(skill_registry.skills.keys()), n=1, cutoff=0.7)找最接近的合法名称。
• 最关键的是：在plan_prompt末尾强制加一句：“可用技能名必须严格从以下列表中选择：['get_weather', 'send_sms']，不允许任何变体。” 这招实测将错误率从 12% 降到 0.3%。

5.2 技能执行超时，但 LLM 却认为“成功”

现象：web_searchskill 设置了 30 秒超时，但 LLM 的总结里写“已搜索完成”，而实际上返回了空结果。
根因：skill 的SkillResult没有区分“业务成功”和“技术成功”。HTTP 超时抛出httpx.TimeoutException，skill 捕获后返回failure，但 LLM 的总结 prompt 没有要求它识别failure状态。
解决方案：

• 统一错误分类：在SkillResult中增加error_type字段，如"network_timeout"、"api_rate_limit"、"validation_error"。
• 强化总结 prompt：在summary_prompt中明确指令：“如果任何步骤返回 ERROR，请在总结开头用【警告】标出，并说明具体原因。”
• 添加兜底校验：在 workflow 引擎中，对failure结果自动注入一条 observation："OBSERVATION: CRITICAL FAILURE in {step.skill_name} - {result.error}"，确保 LLM 无法忽略。

5.3 多技能并发调用时，共享资源冲突

现象：当同时运行 5 个send_smsskill 时，阿里云返回Throttling错误，但日志里只看到一个失败，其他 4 个还在继续发。
根因：技能是无状态的，但下游服务（如短信网关）有全局配额限制。每个 skill 独立调用，不知道兄弟进程的存在。
解决方案：

• 引入分布式限流器：用 Redis 实现令牌桶，所有send_smsskill 调用前先redis.eval(LUA_SCRIPT, 1, "sms_quota", 1)申请令牌。
• 更优雅的做法：在sms_skill内部封装一个SmsRateLimiter单例，用threading.Lock控制本地并发，再配合 Redis 做集群协调。
• 经验之谈：不要在 Agent 层做限流！必须下沉到 skill 内部。因为 Agent 可能被多个实例部署，而 skill 是唯一能感知“自己在做什么”的实体。

5.4 技能输出数据过大，导致 LLM 上下文爆炸

现象：fetch_documentskill 返回一篇 5000 字的 PDF 文本，LLM 的summary_prompt直接超出上下文长度，报错context_length_exceeded。
根因：skill 的职责是“执行”，不是“决策”。它不应该假设 LLM 能处理多大体积的数据。
解决方案：

• 强制输出截断：在 skill 的SkillResult.success(data)前，加入：

  if isinstance(data, str) and len(data) > 2000: data = data[:2000] + "...（内容过长，已截断）"

• 提供多粒度输出：为fetch_documentskill 增加output_level参数："summary"（200字摘要）、"key_points"（5个要点）、"full_text"（全文，但需显式声明风险）。
• 终极方案：用 RAG 替代全文传输。skill 只返回文档 ID 和向量库中的 chunk IDs，LLM 通过retrieve_chunksskill 按需拉取——这才是 scalable 的设计。

5.5 生产环境技能调用链路黑盒，无法快速定位故障点

现象：用户投诉“订会议室失败”，但日志里只有Agent finished with error，找不到是日历查询失败，还是邮件发送失败。
根因：缺乏端到端的 trace ID 和结构化日志。
解决方案：

• 统一 trace ID 注入：在 Agent 入口生成trace_id = str(uuid.uuid4())，并透传给每个 skill 调用。
• 标准化日志格式：所有 skill 日志必须包含trace_id、skill_name、input_hash、duration_ms、status。
• 可视化看板：用 Grafana + Loki 搭建技能健康度看板，监控每个 skill 的 P95 延迟、错误率、成功率趋势。我们曾靠这个看板发现notion_update_page的错误率在每天上午 10 点飙升，最终定位到是 Notion 的 rate limit 重置时间点。

注意：不要用print()或logging.info()打日志！必须用结构化 logger（如structlog），确保每条日志能被机器解析。我们曾因日志格式不统一，导致运维同学花 3 小时手动 grep 200MB 日志文件。

6. 技能治理与演进：如何让 skills 系统持续健康生长

一个 skills 系统上线只是开始，真正的挑战在于长期治理。我们总结了一套“技能健康度”评估框架，每月对所有 skill 进行扫描，指标包括：可用性（uptime）、准确性（output_schema_conformance）、效率（p95_latency）、健壮性（failure_recovery_rate）、可维护性（code_churn_rate）。其中，code_churn_rate（代码变更频率）是个反直觉但极有价值的指标——如果一个 skill 的代码每月修改超过 3 次，说明它的契约（Schema）设计有问题，总在打补丁；健康的 skill 应该稳定运行半年以上，只做小修小补。

6.1 技能版本灰度与回滚机制

当要升级web_searchskill 以支持新搜索引擎时，绝不能一刀切。我们的做法是：

• 双注册模式：skill_registry.register("web_search", new_search_skill, version="v2")和skill_registry.register("web_search", old_search_skill, version="v1")。
• 流量染色：在 Agent 的plan_prompt中加入当前环境标识（env=prod-canary），LLM 生成时带上{"tool": "web_search", "version": "v2"}。
• 自动降级：如果v2的错误率连续 5 分钟超过 5%，系统自动将所有v2请求路由到v1，并触发告警。
  这套机制让我们在某次升级 SerpAPI 到 Google Custom Search 时，零用户感知完成了迁移。

6.2 技能市场与跨团队复用

skills 最大的价值在于复用。我们建立了内部“技能市场”：一个静态网站，展示所有已注册 skill 的文档、Schema、调用示例、SLA 承诺（如get_weatherP95 < 800ms）。每个 skill 的 README.md 都包含：

• 适用场景：如“适用于实时性要求不高的天气概览，不适用于气象预警”。
• 已知限制：如“仅支持中国城市，海外城市需用get_weather_international”。
• 性能基线：表格形式展示不同输入下的耗时（city="北京"vscity="纽约"）。
  这避免了重复造轮子。某业务线曾想开发自己的短信 skill，看到市场里已有send_sms支持阿里云/腾讯云双通道，且 SLA 99.95%，立刻放弃。

6.3 技能的“退休”与“归档”

技术债最大的来源不是写新代码，而是不敢删旧代码。我们规定：

• 退休条件：一个 skill 连续 9