0Pricing
AI Agents · 课时

插件发现与注册

工具注册表、清单文件和运行时动态加载工具

插件发现与注册 是 CoddyKit 上的免费 AI Agents 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

代理的插件系统

插件系统允许代理在运行时发现并加载新工具，而无需修改核心代理代码。代理会读取插件目录，加载每个插件的清单，验证清单，并将其中的工具添加到当前工具注册表中。

这样可以实现模块化且可扩展的代理架构。

插件清单格式

每个插件都随附一个 plugin.json 清单文件。这是插件的身份卡：它是什么、版本是什么、提供哪些工具，以及需要什么（Python 软件包、环境变量）。

# plugin.json — stored in the plugin's root directory
EXAMPLE_MANIFEST = {
    'name': 'weather-tools',
    'display_name': 'Weather Tools',
    'version': '2.1.0',
    'description': 'Real-time weather and forecast tools',
    'author': 'Jane Developer <jane@example.com>',
    'license': 'MIT',
    'entry_point': 'weather_tools.plugin',  # Python module path
    'tool_definitions': [
        'get_current_weather',
        'get_5day_forecast',
        'get_weather_alerts'
    ],
    'requires': {
        'python_packages': ['requests>=2.28'],
        'env_vars': ['WEATHER_API_KEY']
    },
    'tags': ['weather', 'forecast', 'iot'],
    'min_framework_version': '1.0.0'
}

import json
print(json.dumps(EXAMPLE_MANIFEST, indent=2)[:300])

插件目录加载器

请扫描目录中的插件子目录。每个子目录都必须包含一个 plugin.json 清单。加载器会读取每个清单、验证清单，并在插件注册表中注册插件。

import os
import json

PLUGIN_DIR = 'plugins'

def discover_plugins(plugin_dir: str = PLUGIN_DIR) -> list:
    discovered = []
    if not os.path.isdir(plugin_dir):
        print(f'Plugin directory not found: {plugin_dir}')
        return []

    for entry in os.scandir(plugin_dir):
        if not entry.is_dir():
            continue
        manifest_path = os.path.join(entry.path, 'plugin.json')
        if not os.path.exists(manifest_path):
            print(f'Skipping {entry.name}: no plugin.json')
            continue
        try:
            with open(manifest_path) as f:
                manifest = json.load(f)
            manifest['_path'] = entry.path
            manifest['_name'] = entry.name
            discovered.append(manifest)
            print(f'Discovered: {manifest["name"]} v{manifest["version"]}')
        except (json.JSONDecodeError, KeyError) as e:
            print(f'Invalid manifest in {entry.name}: {e}')

    return discovered

plugins = discover_plugins()
print(f'Total plugins discovered: {len(plugins)}')

清单验证

加载插件之前，请根据模式验证其清单，以发现必填字段缺失、版本无效或环境变量缺失等问题。请拒绝无效插件，并记录原因。

import os

REQUIRED_MANIFEST_FIELDS = {'name', 'version', 'entry_point', 'tool_definitions'}

def validate_manifest(manifest: dict) -> tuple:
    """
    Returns (is_valid: bool, errors: list)
    """
    errors = []

    # Required fields
    missing = REQUIRED_MANIFEST_FIELDS - set(manifest.keys())
    if missing:
        errors.append(f'Missing required fields: {missing}')

    # Version format
    version = manifest.get('version', '')
    parts = version.split('.')
    if len(parts) != 3 or not all(p.isdigit() for p in parts):
        errors.append(f'Invalid version format: {version}')

    # Check required env vars exist
    env_vars = manifest.get('requires', {}).get('env_vars', [])
    for var in env_vars:
        if not os.environ.get(var):
            errors.append(f'Missing required env var: {var}')

    # Tool definitions must be a non-empty list
    tools = manifest.get('tool_definitions', [])
    if not isinstance(tools, list) or len(tools) == 0:
        errors.append('tool_definitions must be a non-empty list')

    return len(errors) == 0, errors

if __name__ == '__main__':
    good_manifest = {
        'name': 'weather-tools', 'version': '1.2.0',
        'entry_point': 'weather_tools.plugin', 'tool_definitions': [{'name': 'get_weather'}]
    }
    bad_manifest = {'name': 'broken-plugin', 'version': 'v1'}
    print('Good manifest:', validate_manifest(good_manifest))
    print('Bad manifest: ', validate_manifest(bad_manifest))

动态加载工具

插件清单通过验证后，请动态导入插件的 Python 模块，并调用 get_tools() 来获取工具定义和执行函数。Python 的 importlib 可以轻松完成这一过程。

import importlib
import importlib.util
import sys

def load_plugin_module(manifest: dict):
    """
    Import the plugin's Python module and return it.
    Adds the plugin directory to sys.path if needed.
    """
    plugin_path = manifest['_path']
    entry_point = manifest['entry_point']  # e.g. 'weather_tools.plugin'

    # Add plugin directory to path so relative imports work
    if plugin_path not in sys.path:
        sys.path.insert(0, plugin_path)

    try:
        module = importlib.import_module(entry_point)
        return module
    except ImportError as e:
        print(f'Failed to import {entry_point}: {e}')
        return None

def extract_tools_from_module(module, tool_names: list) -> dict:
    """
    Returns {tool_name: {'schema': dict, 'execute': callable}}
    """
    tools = {}
    for name in tool_names:
        schema_fn = getattr(module, f'get_{name}_schema', None)
        execute_fn = getattr(module, f'execute_{name}', None)
        if schema_fn and execute_fn:
            tools[name] = {'schema': schema_fn(), 'execute': execute_fn}
    return tools

if __name__ == '__main__':
    class FakeModule:
        def get_weather_schema(self):
            return {'name': 'get_weather'}
        def execute_weather(self, params):
            return {'temp': 72}

    tools = extract_tools_from_module(FakeModule(), ['weather'])
    print('Extracted tools:', list(tools.keys()))
    print('Weather schema:', tools['weather']['schema'])

工具注册表

工具注册表是所有可用工具的唯一可信来源。它将工具名称映射到对应的模式和执行函数。代理会在为每次 LLM 调用构建工具列表时查询注册表。

class ToolRegistry:
    def __init__(self):
        self._tools: dict = {}  # name -> {'schema', 'execute', 'plugin'}
        self._plugins: dict = {}  # plugin_name -> manifest

    def register_tool(
        self, name: str, schema: dict,
        execute_fn, plugin_name: str
    ):
        if name in self._tools:
            print(f'WARNING: Tool {name} already registered, overwriting')
        self._tools[name] = {
            'schema': schema,
            'execute': execute_fn,
            'plugin': plugin_name
        }

    def register_plugin(self, manifest: dict, tools: dict):
        self._plugins[manifest['name']] = manifest
        for name, tool in tools.items():
            self.register_tool(name, tool['schema'],
                               tool['execute'], manifest['name'])
        print(f'Registered plugin: {manifest["name"]} '
              f'({len(tools)} tools)')

    def get_all_schemas(self) -> list:
        return [t['schema'] for t in self._tools.values()]

    def execute(self, tool_name: str, params: dict):
        if tool_name not in self._tools:
            raise KeyError(f'Unknown tool: {tool_name}')
        return self._tools[tool_name]['execute'](params)

registry = ToolRegistry()

if __name__ == '__main__':
    def get_weather(params):
        return {'temp': 72, 'city': params.get('city')}

    manifest = {'name': 'weather-tools'}
    tools = {'get_weather': {'schema': {'name': 'get_weather'}, 'execute': get_weather}}
    registry.register_plugin(manifest, tools)
    print('Registered schemas:', registry.get_all_schemas())
    print('Execution result:', registry.execute('get_weather', {'city': 'Paris'}))

按能力搜索

当代理拥有许多插件时，应能够按能力标签搜索注册表，而不是将所有工具都加载到每次 LLM 调用中。上下文中的工具过多会降低 LLM 选择工具的准确性。

class SearchableToolRegistry(ToolRegistry):
    def search(self, query: str) -> list:
        """
        Search tools by name, description, or tags.
        Returns list of matching tool schemas.
        """
        query_lower = query.lower()
        matches = []
        for name, tool in self._tools.items():
            schema = tool['schema']
            plugin = self._plugins.get(tool['plugin'], {})
            plugin_tags = plugin.get('tags', [])

            if (
                query_lower in name.lower() or
                query_lower in schema.get('description', '').lower() or
                any(query_lower in tag.lower() for tag in plugin_tags)
            ):
                matches.append(schema)
        return matches

# Usage:
registry = SearchableToolRegistry()
# After loading plugins...
weather_tools = registry.search('weather')
print(f'Weather tools: {[t["name"] for t in weather_tools]}')

目录变化时热重载

在开发过程中，文件发生变化时无需重启代理即可重新加载插件非常有用。请使用 watchdog 库监视插件目录，并在 plugin.json 文件发生变化时触发重载。

# pip install watchdog
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class PluginReloadHandler(FileSystemEventHandler):
    def __init__(self, registry, plugin_loader_fn):
        self.registry = registry
        self.loader = plugin_loader_fn

    def on_modified(self, event):
        if 'plugin.json' in event.src_path:
            print(f'Plugin manifest changed: {event.src_path}')
            self.reload_plugin(event.src_path)

    def reload_plugin(self, manifest_path: str):
        import json, os
        plugin_dir = os.path.dirname(manifest_path)
        try:
            with open(manifest_path) as f:
                manifest = json.load(f)
            manifest['_path'] = plugin_dir
            self.loader(manifest, self.registry)
            print(f'Reloaded: {manifest["name"]}')
        except Exception as e:
            print(f'Reload failed: {e}')

def start_plugin_watcher(plugin_dir: str, registry):
    handler = PluginReloadHandler(registry, lambda m, r: None)
    observer = Observer()
    observer.schedule(handler, plugin_dir, recursive=True)
    observer.start()
    return observer

依赖检查与自动安装

加载插件时，请检查其所需的 Python 软件包是否已安装。也可以选择使用 pip 自动安装缺失的软件包。如果某个依赖项无法满足，请记录清晰的错误。

import subprocess
import sys
import importlib

def check_and_install_deps(manifest: dict, auto_install: bool = False) -> bool:
    packages = manifest.get('requires', {}).get('python_packages', [])
    missing = []

    for pkg_spec in packages:
        pkg_name = pkg_spec.split('>=')[0].split('==')[0].strip()
        try:
            importlib.import_module(pkg_name.replace('-', '_'))
        except ImportError:
            missing.append(pkg_spec)

    if not missing:
        return True

    print(f'Missing packages for {manifest["name"]}: {missing}')

    if auto_install:
        for pkg in missing:
            print(f'Installing {pkg}...')
            result = subprocess.run(
                [sys.executable, '-m', 'pip', 'install', pkg],
                capture_output=True, text=True
            )
            if result.returncode != 0:
                print(f'Install failed: {result.stderr[:200]}')
                return False
        return True

    return False

if __name__ == '__main__':
    manifest = {'name': 'demo-plugin', 'requires': {'python_packages': ['totally_fake_package_xyz']}}
    ok = check_and_install_deps(manifest, auto_install=False)
    print('All dependencies satisfied:', ok)

完整的插件引导

请将发现、验证、依赖检查、模块加载和注册表注册整合到一个引导函数中，并在代理启动时调用该函数。

def bootstrap_plugins(
    plugin_dir: str,
    registry: ToolRegistry,
    auto_install_deps: bool = False
) -> dict:
    results = {'loaded': [], 'failed': []}
    manifests = discover_plugins(plugin_dir)

    for manifest in manifests:
        name = manifest.get('name', '?')
        is_valid, errors = validate_manifest(manifest)
        if not is_valid:
            print(f'INVALID {name}: {errors}')
            results['failed'].append({'name': name, 'reason': errors})
            continue

        deps_ok = check_and_install_deps(manifest, auto_install_deps)
        if not deps_ok:
            results['failed'].append({'name': name, 'reason': 'missing_deps'})
            continue

        module = load_plugin_module(manifest)
        if not module:
            results['failed'].append({'name': name, 'reason': 'import_error'})
            continue

        tools = extract_tools_from_module(module, manifest['tool_definitions'])
        registry.register_plugin(manifest, tools)
        results['loaded'].append(name)

    print(f'Plugins loaded: {results["loaded"]}')
    print(f'Plugins failed: {results["failed"]}')
    return results

记录插件生命周期事件

请记录插件生命周期中的每个重要事件：发现、验证失败、成功加载、热重载和卸载。这些日志对于在生产环境中调试和审计插件系统至关重要。

import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
plugin_logger = logging.getLogger('plugin_system')

def log_plugin_event(event: str, plugin_name: str, details: dict = None):
    entry = {
        'event': event,
        'plugin': plugin_name,
        'timestamp': datetime.utcnow().isoformat(),
        'details': details or {}
    }
    if event in ('VALIDATION_FAILED', 'LOAD_FAILED', 'SIGNATURE_INVALID'):
        plugin_logger.warning('Plugin event: %s', entry)
    else:
        plugin_logger.info('Plugin event: %s', entry)

# Usage:
log_plugin_event('DISCOVERED', 'weather-tools', {'path': 'plugins/weather-tools'})
log_plugin_event('LOADED', 'weather-tools', {'tools': ['get_current_weather']})
log_plugin_event('VALIDATION_FAILED', 'bad-plugin', {'errors': ['Missing entry_point']})

if __name__ == '__main__':
    import sys
    plugin_logger.addHandler(logging.StreamHandler(sys.stdout))
    log_plugin_event('LOADED', 'demo-plugin', {'tools': ['get_weather']})

知识检查

插件清单中的 entry_point 字段有什么作用？

回顾：插件发现与注册

太好了！您学到了：

  • 插件清单：包含 name、version、entry_point、tool_definitions、requires 的 plugin.json
  • 发现：扫描目录，查找包含 plugin.json 的子目录
  • 验证：检查必填字段、版本格式、环境变量和非空工具列表
  • 动态加载：使用 entry_point 路径调用 importlib.import_module()
  • 工具注册表：将工具名称映射到模式和 execute 函数的中央映射表
  • 热重载：watchdog 监视插件目录，并在清单变化时重新加载

下一课：工具版本管理与兼容性管理。

常见问题解答

「插件发现与注册」课时是免费的吗？

是的 — 「插件发现与注册」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「插件发现与注册」这节课中我会学到什么？

工具注册表、清单文件和运行时动态加载工具 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。

「插件发现与注册」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 设计可共享的智能体工具
  2. 插件发现与注册
  3. 工具版本管理与兼容性
  4. 构建智能体插件市场
← 返回 AI Agents