Nekro Agent 文档站

Appearance

动态包导入 ​

Nekro Agent 提供了动态包导入功能，允许插件在运行时按需安装和导入 Python 第三方包。这个特性让插件能够在不修改系统环境的前提下，灵活地使用各种 Python 生态中的工具库和框架。

功能概述 ​

动态包导入功能具备以下特点：

  • 按需安装：只在需要时才下载和安装包，节省系统资源
  • 版本控制：支持精确的版本规范和依赖约束
  • 隔离存储：包安装在插件专属目录，不影响系统 Python 环境
  • 镜像支持：支持配置 PyPI 镜像源，加速国内下载
  • 错误处理：提供友好的错误提示和异常处理机制

基础用法 ​

导入函数 ​

从 nekro_agent.api.plugin 导入动态包导入函数：

python
from nekro_agent.api.plugin import dynamic_import_pkg
from nekro_agent.api.schemas import AgentCtx

简单导入示例 ​

最简单的用法是直接提供包名：

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "use_requests", "使用 requests 库发送 HTTP 请求")
async def fetch_url(_ctx: AgentCtx, url: str) -> str:
    """使用动态导入的 requests 库获取网页内容"""

    # 动态导入 requests 包
    requests = dynamic_import_pkg("requests")

    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return f"成功获取页面，状态码: {response.status_code}，内容长度: {len(response.text)} 字符"
    except Exception as e:
        return f"请求失败: {e}"

版本规范 ​

指定版本范围 ​

支持标准的 Python 包版本规范语法：

python
# 精确版本
pandas = dynamic_import_pkg("pandas==2.0.0")

# 最低版本要求
numpy = dynamic_import_pkg("numpy>=1.24.0")

# 版本范围
requests = dynamic_import_pkg("requests>=2.25.0,<3.0.0")

# 排除特定版本
flask = dynamic_import_pkg("flask>=2.0,!=2.0.1")

版本规范示例 ​

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "data_analysis", "执行数据分析")
async def analyze_data(_ctx: AgentCtx, data: list) -> str:
    """使用 pandas 进行数据分析"""

    # 导入特定版本的 pandas
    pd = dynamic_import_pkg(
        "pandas>=2.0.0,<3.0.0",
        import_name="pandas"  # 指定导入时的模块名
    )

    try:
        df = pd.DataFrame(data)
        summary = df.describe().to_string()
        return f"数据分析结果：\n{summary}"
    except Exception as e:
        return f"分析失败: {e}"

高级参数配置 ​

完整参数说明 ​

python
dynamic_import_pkg(
    package_spec: str,                    # 包规范，如 "requests>=2.25.0"
    import_name: Optional[str] = None,    # 导入名称，默认使用包名
    mirror: Optional[str] = "https://pypi.tuna.tsinghua.edu.cn/simple",  # PyPI 镜像源
    trusted_host: bool = True,            # 是否信任镜像源主机
    timeout: int = 300,                   # 安装超时时间（秒）
    repo_dir: Optional[Path] = None       # 自定义安装目录
)

使用国内镜像加速 ​

默认使用清华大学 PyPI 镜像源，也可以指定其他镜像：

python
# 使用阿里云镜像
beautifulsoup4 = dynamic_import_pkg(
    "beautifulsoup4",
    mirror="https://mirrors.aliyun.com/pypi/simple/",
    trusted_host=True
)

# 使用腾讯云镜像
pillow = dynamic_import_pkg(
    "Pillow>=10.0.0",
    mirror="https://mirrors.cloud.tencent.com/pypi/simple",
    trusted_host=True
)

# 使用官方 PyPI（较慢）
scipy = dynamic_import_pkg(
    "scipy",
    mirror="https://pypi.org/simple",
    trusted_host=False
)

实用场景示例 ​

示例 1：网页爬取 ​

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "scrape_webpage", "抓取网页内容")
async def scrape_webpage(_ctx: AgentCtx, url: str, selector: str) -> str:
    """使用 BeautifulSoup 抓取网页指定元素"""

    # 动态导入所需的包
    requests = dynamic_import_pkg("requests>=2.25.0")
    bs4 = dynamic_import_pkg("beautifulsoup4>=4.9.0", import_name="bs4")

    try:
        # 获取网页内容
        response = requests.get(url, timeout=10)
        response.raise_for_status()

        # 解析 HTML
        soup = bs4.BeautifulSoup(response.text, 'html.parser')
        elements = soup.select(selector)

        if not elements:
            return f"未找到匹配选择器 '{selector}' 的元素"

        # 提取文本内容
        results = [elem.get_text(strip=True) for elem in elements[:5]]  # 最多返回 5 个
        return f"找到 {len(elements)} 个元素，前 {len(results)} 个内容：\n" + "\n".join(results)

    except Exception as e:
        return f"抓取失败: {e}"

示例 2：图像处理 ​

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "process_image", "处理图像文件")
async def process_image(_ctx: AgentCtx, image_path: str, operation: str) -> str:
    """使用 Pillow 处理图像"""

    # 动态导入 Pillow
    PIL = dynamic_import_pkg("Pillow>=10.0.0", import_name="PIL")
    from PIL import Image, ImageFilter

    try:
        # 获取实际文件路径
        real_path = _ctx.fs.get_file(image_path)

        # 打开图像
        img = Image.open(real_path)

        # 根据操作类型处理
        if operation == "blur":
            processed = img.filter(ImageFilter.BLUR)
        elif operation == "grayscale":
            processed = img.convert('L')
        elif operation == "resize":
            processed = img.resize((800, 600))
        else:
            return f"不支持的操作: {operation}"

        # 保存处理后的图像
        output_path = _ctx.fs.shared_path / f"processed_{operation}.jpg"
        processed.save(output_path)

        # 返回给 AI 的沙盒路径
        sandbox_path = _ctx.fs.forward_file(output_path)
        return f"图像处理完成，结果: {sandbox_path}"

    except Exception as e:
        return f"图像处理失败: {e}"

示例 3：数据处理与可视化 ​

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "create_chart", "创建数据图表")
async def create_data_chart(_ctx: AgentCtx, data: dict, chart_type: str) -> str:
    """使用 matplotlib 创建数据图表"""

    # 动态导入数据处理和可视化库
    pd = dynamic_import_pkg("pandas>=2.0.0", import_name="pandas")
    plt = dynamic_import_pkg("matplotlib>=3.5.0", import_name="matplotlib.pyplot")

    try:
        # 创建 DataFrame
        df = pd.DataFrame(data)

        # 创建图表
        fig, ax = plt.subplots(figsize=(10, 6))

        if chart_type == "bar":
            df.plot(kind='bar', ax=ax)
        elif chart_type == "line":
            df.plot(kind='line', ax=ax)
        elif chart_type == "pie":
            df.plot(kind='pie', y=df.columns[0], ax=ax)
        else:
            return f"不支持的图表类型: {chart_type}"

        # 保存图表
        output_path = _ctx.fs.shared_path / f"chart_{chart_type}.png"
        plt.savefig(output_path, dpi=150, bbox_inches='tight')
        plt.close(fig)

        # 返回沙盒路径
        sandbox_path = _ctx.fs.forward_file(output_path)
        return f"图表已生成: {sandbox_path}"

    except Exception as e:
        return f"图表生成失败: {e}"

示例 4：科学计算 ​

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "scientific_compute", "执行科学计算")
async def scientific_compute(_ctx: AgentCtx, operation: str, values: list) -> str:
    """使用 NumPy 和 SciPy 进行科学计算"""

    # 动态导入科学计算库
    np = dynamic_import_pkg("numpy>=1.24.0", import_name="numpy")
    scipy = dynamic_import_pkg("scipy>=1.10.0")

    try:
        data = np.array(values)

        if operation == "fft":
            # 快速傅里叶变换
            result = np.fft.fft(data)
            return f"FFT 结果（前 5 个）: {result[:5]}"

        elif operation == "stats":
            # 统计分析
            from scipy import stats
            mean = np.mean(data)
            std = np.std(data)
            skew = stats.skew(data)
            kurtosis = stats.kurtosis(data)

            return f"""统计结果:
- 均值: {mean:.4f}
- 标准差: {std:.4f}
- 偏度: {skew:.4f}
- 峰度: {kurtosis:.4f}"""

        elif operation == "interpolate":
            # 数据插值
            from scipy import interpolate
            x = np.arange(len(data))
            f = interpolate.interp1d(x, data, kind='cubic')
            x_new = np.linspace(0, len(data)-1, len(data)*2)
            y_new = f(x_new)
            return f"插值完成，原始数据点: {len(data)}，插值后: {len(y_new)}"

        else:
            return f"不支持的操作: {operation}"

    except Exception as e:
        return f"计算失败: {e}"

错误处理 ​

动态导入可能会因为网络问题、包不存在等原因失败

常见错误类型 ​

python
# RuntimeError: 安装失败
# - 网络连接问题
# - 包不存在或版本不可用
# - SSL 证书验证失败
# - 镜像源访问被拒绝

# ImportError: 导入失败
# - 包已安装但无法找到模块
# - 导入名称与实际模块名不匹配

# ValueError: 包规范格式错误
# - 版本号格式不正确
# - 使用了不支持的版本运算符

# subprocess.TimeoutExpired: 安装超时
# - 包体积过大
# - 网络速度过慢

注意事项 ​

安全性考虑 ​

  1. 仅导入可信的包：只安装来自官方 PyPI 或可信镜像源的包
  2. 版本锁定：在生产环境中使用精确的版本号，避免意外的包更新
  3. 审查依赖：了解包的传递依赖，避免引入不必要的风险
  4. 文档声明：在插件文档中声明使用的包，并注明版本号

性能考虑 ​

  1. 首次安装开销：第一次使用某个包时会有下载和安装时间
  2. 磁盘空间：每个包都会占用磁盘空间，注意控制包的数量和大小

兼容性考虑 ​

  1. Python 版本：确保包支持 Nekro Agent 运行的 Python 版本
  2. 系统依赖：某些包可能需要系统级的依赖库（如 OpenCV 需要特定的 C++ 库）
  3. 包冲突：注意不同包之间可能的依赖冲突

故障排查 ​

安装成功但导入失败 ​

错误：ImportError: 安装成功但无法导入 module_name

解决方案：
1. 检查 import_name 参数是否正确
2. 某些包的导入名与包名不同（如 Pillow -> PIL）
3. 包可能需要额外的系统依赖

通过合理使用动态包导入功能，你的插件可以灵活地利用 Python 丰富的生态系统，实现更加强大和多样化的功能。

发现文档问题？报告问题