详解基于 Python 的 Ollama 自动化部署脚本

在 AI 大模型本地化部署的场景中，自动化脚本能够极大提升效率并减少人为操作失误。本文将详细解析一款基于 Python 的 Ollama 自动化部署脚本，该脚本实现了从服务器 SSH 连接、依赖安装到 Ollama 服务部署及模型拉取的全流程自动化，特别适合需要批量部署或标准化部署流程的场景。

脚本整体架构与核心功能

这款自动化部署脚本采用模块化设计，主要包含三个核心功能模块：

• SSH 连接管理模块（带重试机制）
• 远程命令执行模块（支持输出解析与重试）
• 主流程控制模块（串联整个部署流程）

通过这三个模块的协作，脚本能够完成从服务器连接到最终模型可用的全流程操作，支持网络波动容错、命令执行超时处理和关键步骤重试机制。

核心模块详解

1. SSH 连接管理模块（connect_with_retry 函数）

该模块负责建立与目标服务器的 SSH 连接，解决了网络不稳定情况下的连接可靠性问题。

def connect_with_retry(host, port, username, password, max_retries=3, retry_delay=10):
    """带重试机制的SSH连接"""
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())  # 自动接受未知主机密钥
    
    for attempt in range(max_retries):
        try:
            print(f"连接到 {host}:{port} (第 {attempt + 1}/{max_retries} 次尝试)...")
            ssh.connect(
                host,
                port=port,
                username=username,
                password=password,
                timeout=30,
                banner_timeout=30,
                auth_timeout=30
            )
            print("连接成功")
            return ssh
        except socket.timeout:
            print(f"连接超时 (第 {attempt + 1} 次尝试失败)")
        except paramiko.AuthenticationException:
            print("认证失败，请检查用户名和密码")
            return None
        except Exception as e:
            print(f"连接失败: {str(e)} (第 {attempt + 1} 次尝试失败)")
        
        if attempt < max_retries - 1:
            print(f"将在 {retry_delay} 秒后重试...")
            time.sleep(retry_delay)
    
    print(f"超过最大重试次数 ({max_retries} 次)，连接失败")
    return None

核心特性解析：

• 采用重试机制（默认 3 次），应对临时网络波动
• 细分错误类型：超时、认证失败和通用异常，提供精准错误提示
• 指数退避策略：失败后等待一定时间再重试，避免网络拥塞加剧
• 自动接受主机密钥：适合初次连接场景，简化部署流程

使用场景：在服务器负载较高或网络质量不稳定时，该函数能显著提高连接成功率，减少人工干预需求。

2. 远程命令执行模块（execute_command 函数）

该模块是脚本的核心，负责在远程服务器上执行命令，处理输出解析、进度展示和错误重试。

python

运行

def execute_command(ssh, command, timeout=600, print_output=True, retry=1):
    """执行命令并返回输出结果，支持重试机制"""
    for attempt in range(retry + 1):
        try:
            stdin, stdout, stderr = ssh.exec_command(command, get_pty=True)
            stdin.close()

            output = []
            error = []
            end_time = time.time() + timeout
            last_progress = -1  # 用于跟踪进度百分比

            # 编译进度匹配正则表达式
            progress_pattern = re.compile(r'(\d+(\.\d+)?)%')

            while not stdout.channel.exit_status_ready():
                if time.time() > end_time:
                    raise TimeoutError(f"命令执行超时: {command}")

                # 处理标准输出
                if stdout.channel.recv_ready():
                    line_bytes = stdout.readline()
                    # 字节转字符串处理（支持多编码）
                    if isinstance(line_bytes, bytes):
                        try:
                            line = line_bytes.decode('utf-8').strip()
                        except UnicodeDecodeError:
                            line = line_bytes.decode('latin-1').strip()
                    else:
                        line = line_bytes.strip()

                    if line:
                        output.append(line)
                        if print_output:
                            # 进度信息处理
                            if 'pulling' in line or '%' in line:
                                progress_match = progress_pattern.search(line)
                                if progress_match:
                                    progress = float(progress_match.group(1))
                                    # 进度变化超过1%才显示，减少冗余输出
                                    if progress - last_progress >= 1 or progress == 100:
                                        last_progress = progress
                                        print(f"进度: {progress:.1f}%")
                            # 关键状态信息处理
                            elif any(keyword in line for keyword in
                                     ['>>>', 'Complete', 'Installing', 'Downloading', 'Created symlink', 'WARNING',
                                      'error']):
                                clean_line = re.sub(r'[\r\x1b\[0-9;]*[m\x0f]', '', line)
                                print(f"输出: {clean_line}")
                            # 有意义的普通输出处理
                            elif len(line) > 5 and not re.match(r'^[#= ]+$', line):
                                clean_line = re.sub(r'[\r#= ]{5,}', '', line)
                                if clean_line:
                                    print(f"输出: {clean_line}")

                # 错误输出处理（略）
                # ...

            exit_status = stdout.channel.recv_exit_status()

            # 重试逻辑处理
            if exit_status == 0:
                return exit_status, output, error
            elif attempt < retry:
                print(f"命令执行失败，将在5秒后重试 (第 {attempt + 2}/{retry + 1} 次尝试)")
                time.sleep(5)
                continue
            else:
                return exit_status, output, error

        except Exception as e:
            # 异常重试逻辑（略）
            # ...

核心特性解析：

• 智能输出处理：

  • 进度信息过滤：只在进度变化超过 1% 时显示，减少日志冗余
  • 特殊字符清理：移除 ANSI 控制码和多余符号，使输出更易读
  • 多编码支持：同时支持 utf-8 和 latin-1 编码，解决中文乱码问题

• 错误处理机制：

  • 超时控制：默认 10 分钟超时，防止命令无限挂起
  • 分类重试：针对不同命令设置不同重试次数（网络相关命令重试更多）
  • 异常捕获：全面捕获执行过程中的各类异常，保证脚本稳定性

• 进度可视化：通过正则表达式提取进度百分比，提供直观的下载 / 安装进度反馈

使用场景：该函数特别适合处理网络下载（如 Ollama 安装包下载、模型拉取）等耗时操作，通过进度展示和重试机制提升用户体验和成功率。

3. 主流程控制模块（main 函数）

主函数负责串联整个部署流程，定义部署步骤和参数配置。

def main():
    # 服务器连接信息
    host = "88.88.88.65"
    port = 22
    username = "root"
    password = "Admin@2025"

    # 连接服务器（带重试）
    ssh = connect_with_retry(host, port, username, password)
    if not ssh:
        print("无法建立SSH连接，退出脚本")
        return

    try:
        # 安装必要依赖
        print("安装必要依赖...")
        commands = [
            ("yum install -y --allowerasing curl wget", 1),
            ("curl -fsSL https://ollama.com/install.sh | sh", 3),  # 增加重试次数
            ("systemctl start ollama", 1),
            ("systemctl enable ollama", 1),
            ("sleep 10", 1),
            ("ollama pull deepseek-r1:8b", 3),  # 增加重试次数
            ("ollama list", 1)
        ]

        for cmd, retry in commands:
            print(f"\n执行命令: {cmd}")
            exit_status, output, error = execute_command(ssh, cmd, retry=retry)

            if exit_status != 0:
                print(f"命令执行失败: {cmd}")
                print(f"错误信息: {''.join(error)}")
                return

        print("\n所有操作完成！ollama已安装并成功拉取deepseek-r1:8b镜像")

    except Exception as e:
        print(f"发生错误: {str(e)}")
    finally:
        # 关闭连接
        if ssh.get_transport() and ssh.get_transport().is_active():
            ssh.close()
            print("连接已关闭")

核心流程解析：

1. 连接阶段：调用connect_with_retry建立 SSH 连接
2. 部署阶段：按顺序执行预定义命令列表，包括：
   • 基础依赖安装（curl、wget）
   • Ollama 安装脚本执行（带 3 次重试）
   • 服务启动与开机自启配置
   • 模型拉取（带 3 次重试）
   • 安装结果验证
3. 收尾阶段：确保 SSH 连接正确关闭，释放资源

命令设计思路：

• 对网络敏感操作（如curl下载、ollama pull）设置更多重试次数
• 关键步骤间添加等待时间（sleep 10），确保服务稳定启动
• 每个命令执行结果实时校验，失败立即终止并提示

脚本优势与适用场景

核心优势

1. 高可靠性：通过多层重试机制（连接重试、命令执行重试）应对网络波动和临时故障
2. 清晰反馈：进度可视化和结构化输出，让用户实时了解部署状态
3. 异常处理：完善的错误捕获和提示机制，便于问题定位
4. 通用性强：适用于任何支持 SSH 的 Linux 服务器，稍作修改即可适配不同环境

适用场景

• 企业内部 AI 模型服务批量部署
• 开发 / 测试环境标准化搭建
• 远程服务器无人值守部署
• 教学场景中的快速环境准备

使用与扩展建议

1. 参数化配置：建议将服务器信息（IP、用户名、密码）改为配置文件或命令行参数输入，提高安全性和灵活性
2. 日志增强：可添加日志文件输出功能，方便后续审计和问题排查
3. 更多模型支持：扩展命令列表，支持多种模型的自动拉取
4. 环境检查：增加前置检查步骤（如服务器硬件配置、操作系统版本），提前发现不兼容问题
5. 安全加固：建议使用 SSH 密钥认证替代密码认证，提高安全性

总结

这款 Ollama 自动化部署脚本通过模块化设计和健壮的错误处理机制，实现了从服务器连接到模型可用的全流程自动化。其核心价值在于：

• 降低部署门槛：无需手动执行复杂命令
• 提高部署一致性：标准化流程避免人为操作差异
• 增强容错能力：应对网络波动等常见问题

无论是个人开发者还是企业 IT 团队，都可以基于此脚本构建更完善的 AI 模型部署解决方案，快速实现本地化大模型服务的搭建与上线。