第 58 章

浏览器扩展开发:Manifest V3 + Service Worker 安全调用 Claude API

第五十八章:从 Skill 到 Plugin:升级路径与能力跃迁

58.1 Skill 与 Plugin 的本质差异

在 Claude 生态中,Skill 和 Plugin 常被混用,但它们代表着不同的能力层次和实现复杂度。理解这一差异,是规划升级路径的前提。

Skill 是一种轻量级、无状态的能力扩展。它本质上是一组精心设计的 prompt 模板,告诉 Claude 如何处理某一类特定任务——翻译、代码审查、情感分析、格式转换。Skill 不调用外部 API,不持久化数据,所有逻辑都在 Claude 的推理过程中完成。

Skill 特征:
- 无外部 API 调用
- 无状态(不在请求间持久化数据)
- 纯 prompt 驱动
- 快速部署,无需基础设施
- 能力边界:Claude 的语言理解与生成能力

Plugin 是具备工具调用能力的重量级扩展。它为 Claude 配备了真实的工具——可以查询数据库、调用 REST API、读写文件、执行代码。Plugin 通过 tools 参数注入到 API 调用中,Claude 在推理过程中决定何时调用哪个工具,并将工具结果纳入上下文继续推理。

Plugin 特征:
- 可调用任意外部系统
- 有状态(工具可持久化数据)
- 工具 Schema + 执行逻辑
- 需要服务器/函数计算基础设施
- 能力边界:任何可通过 API 访问的系统

当你的 Skill 开始遇到"我需要实时数据"、"我需要保存结果"、"我需要与外部系统交互"这类需求时,就是升级到 Plugin 的信号。

58.2 升级前的能力评估矩阵

不是所有 Skill 都需要升级成 Plugin。在决策之前,用以下矩阵评估你的场景:

评估维度 适合 Skill 适合 Plugin
数据来源 Claude 的训练知识 实时或私有数据
操作类型 纯分析/生成 有副作用的操作
状态需求 无需跨对话保持状态 需要读写持久化存储
准确性要求 接受 LLM 的概率性误差 需要精确的结构化数据
延迟敏感度 可接受 LLM 推理延迟 对延迟极度敏感(考虑缓存)
维护成本 低(只有 prompt) 高(需要维护工具服务)

典型的升级触发点:

58.3 渐进式升级路径

从 Skill 到 Plugin 的升级不必是一步到位的"大重构"。推荐采用渐进式路径,降低风险。

阶段一:Skill + 手动工具(验证需求)

在真正实现 Plugin 之前,先用"伪 Plugin"验证需求:在 system prompt 中让 Claude 输出结构化的"工具调用指令",由前端代码解析并手动执行。

# 阶段一:伪 Plugin 验证
import anthropic
import json

client = anthropic.Anthropic()

SYSTEM_PROMPT = """你是一个订单助手。当用户询问订单信息时,
请输出以下格式的 JSON,然后我会帮你查询:

<tool_call>
{
  "action": "query_order",
  "order_id": "用户提供的订单号"
}
</tool_call>

获取结果后,我会将结果提供给你,然后你再给出最终回复。"""

def pseudo_plugin_flow(user_message: str, mock_order_data: dict) -> str:
    """第一轮:让 Claude 输出工具调用指令"""
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=512,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_message}]
    )
    
    text = response.content[0].text
    
    # 解析工具调用
    if "<tool_call>" in text:
        start = text.index("<tool_call>") + len("<tool_call>")
        end = text.index("</tool_call>")
        tool_call = json.loads(text[start:end].strip())
        
        # 手动执行(此处用 mock 数据)
        result = mock_order_data.get(tool_call["order_id"], "订单不存在")
        
        # 第二轮:将结果提供给 Claude
        final_response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=512,
            system=SYSTEM_PROMPT,
            messages=[
                {"role": "user", "content": user_message},
                {"role": "assistant", "content": text},
                {"role": "user", "content": f"查询结果:{result}"}
            ]
        )
        return final_response.content[0].text
    
    return text

这个阶段的目的是验证业务逻辑,不是追求工程质量。如果用户确实需要这个功能,再进入阶段二。

阶段二:原生 Tool Use(真正的 Plugin 化)

# 阶段二:原生 Tool Use
import anthropic
from database import get_order_by_id

client = anthropic.Anthropic()

# 将伪 Plugin 中的 action 转化为正式的工具 schema
TOOLS = [
    {
        "name": "query_order",
        "description": "根据订单 ID 查询订单详情",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "订单唯一标识符"
                }
            },
            "required": ["order_id"]
        }
    }
]

def tool_use_flow(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=1024,
            tools=TOOLS,
            messages=messages
        )
        
        if response.stop_reason == "end_turn":
            return response.content[0].text
        
        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    # 调用真实的数据库函数
                    result = get_order_by_id(block.input["order_id"])
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result, ensure_ascii=False)
                    })
            
            messages.extend([
                {"role": "assistant", "content": response.content},
                {"role": "user", "content": tool_results}
            ])

注意阶段一到阶段二的关键变化:

阶段三:生产化加固

原生 Tool Use 跑通之后,还需要一系列生产化工作:

# 阶段三:生产化的 Plugin
import anthropic
import logging
import time
from typing import Optional
from functools import wraps

logger = logging.getLogger(__name__)

def with_retry(max_attempts: int = 3, delay: float = 1.0):
    """工具调用重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    if attempt < max_attempts - 1:
                        logger.warning(f"工具调用失败(第 {attempt+1} 次): {e}")
                        time.sleep(delay * (2 ** attempt))  # 指数退避
            raise last_error
        return wrapper
    return decorator

def with_timeout(seconds: float):
    """工具调用超时控制"""
    import signal
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            def timeout_handler(signum, frame):
                raise TimeoutError(f"工具调用超时({seconds}s)")
            signal.signal(signal.SIGALRM, timeout_handler)
            signal.alarm(int(seconds))
            try:
                return func(*args, **kwargs)
            finally:
                signal.alarm(0)
        return wrapper
    return decorator

class ProductionPlugin:
    """生产化的 Plugin 封装"""
    
    def __init__(self, tools: list, tool_handlers: dict):
        self.client = anthropic.Anthropic()
        self.tools = tools
        self.tool_handlers = tool_handlers
        self.max_iterations = 10  # 防止无限循环
    
    @with_retry(max_attempts=3)
    @with_timeout(seconds=30)
    def _call_tool(self, tool_name: str, tool_input: dict) -> str:
        """调用工具并处理错误"""
        handler = self.tool_handlers.get(tool_name)
        if not handler:
            raise ValueError(f"未知工具: {tool_name}")
        
        try:
            result = handler(**tool_input)
            return json.dumps(result, ensure_ascii=False)
        except Exception as e:
            logger.error(f"工具 {tool_name} 执行失败: {e}", exc_info=True)
            # 返回错误信息而非抛出异常,让 Claude 知道工具失败了
            return json.dumps({"error": str(e), "tool": tool_name})
    
    def run(self, user_message: str, system: Optional[str] = None) -> dict:
        """执行完整的 Plugin 流程,返回详细结果"""
        messages = [{"role": "user", "content": user_message}]
        iterations = 0
        all_tool_calls = []
        
        while iterations < self.max_iterations:
            iterations += 1
            
            response = self.client.messages.create(
                model="claude-opus-4-5",
                max_tokens=2048,
                system=system or "",
                tools=self.tools,
                messages=messages
            )
            
            if response.stop_reason == "end_turn":
                final_text = next(
                    (b.text for b in response.content if hasattr(b, "text")), ""
                )
                return {
                    "response": final_text,
                    "tool_calls": all_tool_calls,
                    "iterations": iterations
                }
            
            if response.stop_reason == "tool_use":
                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        start_time = time.time()
                        result_str = self._call_tool(block.name, block.input)
                        elapsed = time.time() - start_time
                        
                        all_tool_calls.append({
                            "name": block.name,
                            "input": block.input,
                            "elapsed_ms": int(elapsed * 1000)
                        })
                        
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result_str
                        })
                
                messages.extend([
                    {"role": "assistant", "content": response.content},
                    {"role": "user", "content": tool_results}
                ])
        
        raise RuntimeError(f"超过最大迭代次数 ({self.max_iterations})")

58.4 能力跃迁:Plugin 解锁的新范式

从 Skill 升级到 Plugin 不仅仅是"加了工具调用",而是解锁了一系列全新的交互范式。

58.4.1 跃迁一:从静态知识到实时感知

# Plugin 使 Claude 能感知实时世界
REALTIME_TOOLS = [
    {
        "name": "get_stock_price",
        "description": "获取指定股票的实时价格",
        "input_schema": {
            "type": "object",
            "properties": {
                "symbol": {"type": "string", "description": "股票代码,如 AAPL"},
                "currency": {"type": "string", "enum": ["USD", "HKD", "CNY"], "default": "USD"}
            },
            "required": ["symbol"]
        }
    },
    {
        "name": "get_news",
        "description": "获取某个主题的最新新闻",
        "input_schema": {
            "type": "object",
            "properties": {
                "topic": {"type": "string"},
                "hours_ago": {"type": "integer", "description": "获取过去 N 小时的新闻", "default": 24}
            },
            "required": ["topic"]
        }
    }
]

58.4.2 跃迁二:从一次性分析到持续跟踪

# Plugin 支持跨会话的状态持久化
TRACKING_TOOLS = [
    {
        "name": "create_tracker",
        "description": "创建一个持续跟踪任务",
        "input_schema": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "condition": {"type": "string", "description": "触发条件的自然语言描述"},
                "notify_channel": {"type": "string", "enum": ["email", "slack", "webhook"]}
            },
            "required": ["name", "condition", "notify_channel"]
        }
    },
    {
        "name": "get_tracker_history",
        "description": "查看跟踪器的历史触发记录",
        "input_schema": {
            "type": "object",
            "properties": {
                "tracker_id": {"type": "string"},
                "limit": {"type": "integer", "default": 10}
            },
            "required": ["tracker_id"]
        }
    }
]

58.4.3 跃迁三:从被动响应到主动执行

最强大的能力跃迁是 Plugin 使 Claude 从"回答问题的助手"变成"执行任务的代理":

# 多工具协作的复杂任务 Plugin
AGENT_TOOLS = [
    {
        "name": "web_search",
        "description": "搜索网络上的信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "num_results": {"type": "integer", "default": 5}
            },
            "required": ["query"]
        }
    },
    {
        "name": "read_url",
        "description": "读取指定 URL 的内容",
        "input_schema": {
            "type": "object",
            "properties": {
                "url": {"type": "string", "format": "uri"}
            },
            "required": ["url"]
        }
    },
    {
        "name": "write_document",
        "description": "将内容写入文档并保存",
        "input_schema": {
            "type": "object",
            "properties": {
                "title": {"type": "string"},
                "content": {"type": "string"},
                "format": {"type": "string", "enum": ["markdown", "pdf", "docx"]}
            },
            "required": ["title", "content"]
        }
    },
    {
        "name": "send_email",
        "description": "发送电子邮件",
        "input_schema": {
            "type": "object",
            "properties": {
                "to": {"type": "string", "format": "email"},
                "subject": {"type": "string"},
                "body": {"type": "string"},
                "attachment_id": {"type": "string", "description": "write_document 返回的文档 ID"}
            },
            "required": ["to", "subject", "body"]
        }
    }
]

# 用户只需说:"帮我研究一下竞争对手 XYZ 公司最近的动态,写成报告发给老板"
# Claude 会自动:搜索 → 读取多个页面 → 整合信息 → 写文档 → 发邮件

58.5 升级过程中的常见陷阱

陷阱一:工具粒度设计错误

# 错误:工具粒度太粗(一个工具做太多事)
{
    "name": "handle_customer_request",
    "description": "处理客户请求,包括查询订单、修改地址、申请退款等",
    # 这会让 Claude 无法精确选择正确的操作
}

# 正确:每个工具有单一职责
[
    {"name": "get_order", "description": "查询订单详情"},
    {"name": "update_shipping_address", "description": "修改收货地址"},
    {"name": "create_refund_request", "description": "申请退款"}
]

陷阱二:忽略工具的幂等性设计

# 危险:不幂等的工具在重试时会造成重复执行
def send_notification(user_id: str, message: str) -> dict:
    # 如果 Claude 因某种原因重试调用,会发出两条通知!
    notification_service.send(user_id, message)
    return {"sent": True}

# 安全:幂等设计(使用请求 ID)
def send_notification(user_id: str, message: str, idempotency_key: str) -> dict:
    if notification_service.already_sent(idempotency_key):
        return {"sent": True, "duplicate": True}
    notification_service.send(user_id, message, key=idempotency_key)
    return {"sent": True, "duplicate": False}

陷阱三:不处理工具失败的情况

# Claude 需要知道工具失败了,而不是收到空结果
def execute_tool_safely(tool_name: str, tool_input: dict) -> str:
    try:
        result = tool_handlers[tool_name](**tool_input)
        return json.dumps({"success": True, "data": result})
    except PermissionError as e:
        # 告知 Claude 权限不足,让它告知用户
        return json.dumps({"success": False, "error": "permission_denied", "message": str(e)})
    except ValueError as e:
        # 参数错误
        return json.dumps({"success": False, "error": "invalid_input", "message": str(e)})
    except Exception as e:
        # 通用错误
        logger.error(f"工具 {tool_name} 发生意外错误", exc_info=True)
        return json.dumps({"success": False, "error": "internal_error", 
                          "message": "操作暂时不可用,请稍后重试"})

58.6 从单一 Plugin 到 Plugin 生态

成熟的 Plugin 架构不是一个巨大的单体工具集,而是多个专注、可复用的 Plugin 模块组合。

# Plugin 注册中心:动态加载工具
class PluginRegistry:
    def __init__(self):
        self._plugins: dict[str, dict] = {}
    
    def register(self, plugin_id: str, tools: list, handlers: dict, 
                 permissions: list[str] = None):
        """注册一个 Plugin 模块"""
        self._plugins[plugin_id] = {
            "tools": tools,
            "handlers": handlers,
            "permissions": permissions or []
        }
    
    def get_tools_for_user(self, user_id: str, user_roles: list[str]) -> tuple:
        """根据用户权限动态返回可用工具集"""
        available_tools = []
        available_handlers = {}
        
        for plugin_id, plugin in self._plugins.items():
            required = set(plugin["permissions"])
            if not required or required.intersection(user_roles):
                available_tools.extend(plugin["tools"])
                available_handlers.update(plugin["handlers"])
        
        return available_tools, available_handlers

# 使用示例
registry = PluginRegistry()
registry.register("order", ORDER_TOOLS, ORDER_HANDLERS, permissions=["customer_service"])
registry.register("analytics", ANALYTICS_TOOLS, ANALYTICS_HANDLERS, permissions=["analyst", "admin"])
registry.register("public_search", SEARCH_TOOLS, SEARCH_HANDLERS, permissions=[])  # 所有人可用

# 根据用户角色动态组合工具集
tools, handlers = registry.get_tools_for_user("user_001", ["customer_service"])
plugin = ProductionPlugin(tools=tools, tool_handlers=handlers)

小结

从 Skill 到 Plugin 的升级是 Claude 应用从"智能文本处理"向"真实系统代理"的关键跃迁。升级路径建议遵循三阶段:伪 Plugin 验证需求、原生 Tool Use 实现、生产化加固。能力跃迁的三个维度是:静态知识到实时感知、一次性分析到持续跟踪、被动响应到主动执行。关键设计原则包括:工具单一职责、幂等性保证、显式错误传递。当单一 Plugin 无法满足复杂场景时,Plugin 注册中心模式可以实现基于权限的动态工具组合,构建可扩展的 Plugin 生态。

本章评分
4.5  / 5  (3 评分)

💬 留言讨论