当Ollama意外退出后自动恢复推理进度的检查点机制实现思路

Yuan Han 2026-08-06 15:25

一、问题背景：为什么需要Ollama的自动恢复功能

很多人用Ollama做长文本推理的时候都遇到过糟心事：花了大半天写的AI生成小说、做的代码重构、跑的复杂数据分析，突然电脑卡了、Ollama进程崩了，之前跑的内容全没了，只能从头再来。这不是Ollama的锅，而是它本身没做「中途保存进度」的设计——就像写Word没开自动保存，电脑断电就得重写，太闹心。

这时候就需要一个「检查点机制」：简单说就是让Ollama每隔一段时间把当前推理的状态存下来，万一它崩了，下次重启能从上次存的位置接着跑，不用从头开始。

二、核心实现思路：检查点机制的底层逻辑

检查点机制的核心是「存状态+续状态」两步，整个流程不用改Ollama的源码，靠额外的脚本就能实现，门槛很低。

2.1 核心原理拆解

状态是什么？推理过程中，Ollama会不断生成token（可以理解为AI的「思考碎片」），我们要存的就是「已经生成了多少token、生成的内容是什么、推理用到的参数（比如温度、最大生成长度）」。
怎么存？每隔一段时间（比如每生成50个token），把当前的状态存成一个小文件，这个文件就是「检查点」。
怎么续？下次Ollama启动时，先找有没有检查点文件：有就从检查点的位置接着跑，没有就从头开始。

三、具体实现步骤（附完整示例）

整个实现用Python脚本完成，所有代码都能直接复制运行，不用复杂配置。

3.1 前置准备：先搭基础环境

先装两个必须的工具：

Ollama本身：去官网下载对应系统的版本，装完后开一个终端运行ollama serve启动服务（别关这个终端，它是后台服务）。
Python依赖：装Ollama的Python SDK，用来和Ollama交互：

# 装Ollama的Python SDK，和requests（用来发HTTP请求）
pip install ollama requests

3.2 完整实现代码（Python）

整个脚本分三部分：检查点的存储逻辑、推理逻辑、崩溃恢复逻辑。代码里加了详细注释，新手也能看懂。

# 技术栈：Python 3.8+ + Ollama Python SDK + 本地文件存储
import ollama
import os
import json
import time

# -------------------------- 配置参数（可以自己改） --------------------------
MODEL_NAME = "llama3.2:7b"  # 你用的Ollama模型名，比如llama3.2、qwen2都可以
CHECKPOINT_DIR = "./ollama_checkpoints"  # 检查点文件存的文件夹
SAVE_INTERVAL = 50  # 每生成50个token存一次检查点（可以调大，比如100）
MAX_TOKENS = 1000  # 总共要生成的token数（比如写小说要1000字左右）
PROMPT = "写一篇关于乡村振兴的短篇小说，主角是一个回村的大学生，内容要温暖有细节"  # 你的推理提示词
# ---------------------------------------------------------------------------

# 第一步：初始化检查点文件夹，不存在就创建
os.makedirs(CHECKPOINT_DIR, exist_ok=True)

# 第二步：定义检查点的存储和读取函数
def save_checkpoint(generated_tokens: int, generated_content: str, model_name: str, prompt: str):
    """
    把当前推理状态存成检查点文件
    :param generated_tokens: 已经生成的token数
    :param generated_content: 已经生成的内容
    :param model_name: 用的模型名
    :param prompt: 初始提示词
    """
    # 检查点文件的名字用提示词的前20个字符+生成的token数，避免重名
    checkpoint_name = f"{prompt[:20].replace(' ', '_')}_{generated_tokens}.json"
    checkpoint_path = os.path.join(CHECKPOINT_DIR, checkpoint_name)
    
    # 把状态存成JSON格式，方便读取
    checkpoint_data = {
        "generated_tokens": generated_tokens,
        "generated_content": generated_content,
        "model_name": model_name,
        "prompt": prompt,
        "save_time": time.strftime("%Y-%m-%d %H:%M:%S")  # 存的时间，方便排查
    }
    
    with open(checkpoint_path, "w", encoding="utf-8") as f:
        json.dump(checkpoint_data, f, ensure_ascii=False, indent=2)
    print(f"检查点已保存：{checkpoint_path}")

def load_latest_checkpoint(prompt: str):
    """
    读取最新的检查点文件，没有的话返回None
    :param prompt: 初始提示词，用来匹配检查点（避免不同提示词的检查点混了）
    """
    # 先找所有和当前提示词匹配的检查点文件
    prompt_prefix = prompt[:20].replace(' ', '_')
    checkpoint_files = [
        f for f in os.listdir(CHECKPOINT_DIR)
        if f.startswith(prompt_prefix) and f.endswith(".json")
    ]
    
    if not checkpoint_files:
        return None  # 没有检查点，从头开始
    
    # 找到生成token数最多的那个检查点（就是最新的）
    checkpoint_files.sort(key=lambda x: int(x.split("_")[-1].split(".")[0]), reverse=True)
    latest_checkpoint = checkpoint_files[0]
    latest_path = os.path.join(CHECKPOINT_DIR, latest_checkpoint)
    
    # 读取检查点内容
    with open(latest_path, "r", encoding="utf-8") as f:
        checkpoint_data = json.load(f)
    print(f"已加载检查点：{latest_path}，已生成{checkpoint_data['generated_tokens']}个token")
    return checkpoint_data

# 第三步：推理逻辑，分两种情况：从头开始/从检查点续
def run_inference():
    # 先找有没有检查点
    checkpoint = load_latest_checkpoint(PROMPT)
    
    if checkpoint:
        # 有检查点，从检查点的位置接着跑
        current_tokens = checkpoint["generated_tokens"]
        current_content = checkpoint["generated_content"]
        # 剩下要生成的token数：总共要的减去已经生成的
        remaining_tokens = MAX_TOKENS - current_tokens
        print(f"从检查点恢复，还需生成{remaining_tokens}个token")
    else:
        # 没有检查点，从头开始
        current_tokens = 0
        current_content = ""
        remaining_tokens = MAX_TOKENS
        print("无检查点，从头开始推理")
    
    # 如果已经生成的token数够了，就不用跑了
    if remaining_tokens <= 0:
        print("推理已完成，内容如下：")
        print(current_content)
        return
    
    # 开始推理，用Ollama的stream模式（逐token生成，方便计数和存检查点）
    # stream模式会一个一个返回生成的token，比一次性返回更适合存检查点
    response = ollama.generate(
        model=MODEL_NAME,
        prompt=PROMPT,
        options={
            "num_predict": remaining_tokens,  # 这次要生成的token数
            "stream": True  # 开启逐token返回
        }
    )
    
    new_tokens = 0  # 这次新生成的token数
    new_content = ""  # 这次新生成的内容
    
    for chunk in response:
        # 每个chunk是一个token（或者一小段内容）
        token = chunk["response"]
        new_content += token
        new_tokens += 1
        
        # 每生成SAVE_INTERVAL个token，存一次检查点
        if new_tokens % SAVE_INTERVAL == 0:
            total_tokens = current_tokens + new_tokens
            total_content = current_content + new_content
            save_checkpoint(total_tokens, total_content, MODEL_NAME, PROMPT)
    
    # 推理结束，保存最终的检查点（或者直接输出结果）
    total_tokens = current_tokens + new_tokens
    total_content = current_content + new_content
    print("\n推理完成，总token数：", total_tokens)
    print("最终内容：")
    print(total_content)
    # 可以选择删除检查点，或者保留（方便下次继续扩展）
    # save_checkpoint(total_tokens, total_content, MODEL_NAME, PROMPT)

# 主函数，运行推理
if __name__ == "__main__":
    run_inference()

3.3 怎么测试这个脚本？

先开一个终端运行ollama serve（启动Ollama服务）。
再开一个终端，把上面的代码存成ollama_checkpoint.py，运行：

python ollama_checkpoint.py

运行过程中，你可以故意关掉Ollama服务的终端（模拟Ollama意外退出），过几秒再重新开ollama serve，再运行一次python ollama_checkpoint.py，你会发现它会从上次存的位置接着跑，不用从头开始。

四、检查点机制的优化方向

上面的代码是基础版，还有几个可以优化的地方，适合进阶玩家：

4.1 自动清理旧检查点

如果跑的时间长，检查点文件会越来越多，占空间。可以加个逻辑：每次存新检查点的时候，删除旧的检查点（只保留最新的那个）。

# 在save_checkpoint函数最后加这段代码，只保留最新的检查点
def save_checkpoint(...):
    # ... 原来的代码 ...
    # 只保留最新的检查点，删除旧的
    prompt_prefix = prompt[:20].replace(' ', '_')
    all_checkpoints = [
        f for f in os.listdir(CHECKPOINT_DIR)
        if f.startswith(prompt_prefix) and f.endswith(".json")
    ]
    # 排序，把最新的排最后，删除前面的旧文件
    all_checkpoints.sort(key=lambda x: int(x.split("_")[-1].split(".")[0]))
    for old_file in all_checkpoints[:-1]:
        os.remove(os.path.join(CHECKPOINT_DIR, old_file))
        print(f"已删除旧检查点：{old_file}")

4.2 加密检查点文件

如果推理的内容是敏感的（比如商业秘密、个人隐私），可以把检查点的内容加密再存，比如用Python的cryptography库。

4.3 定时检查Ollama状态

上面的脚本是每次重启才检查检查点，还可以加个逻辑：每隔1分钟检查Ollama服务是否正常，一旦发现Ollama崩了，自动保存当前的检查点，然后等Ollama恢复后自动接着跑，不用手动重启脚本。

五、这个机制的应用场景、优缺点和注意事项

5.1 应用场景

长文本生成：比如写长篇小说、论文、代码重构（可能需要跑几小时）。
复杂数据分析：比如用大模型分析几十万字的文档，推理时间很长。
批量推理：比如同时给100篇文章做摘要，万一中途崩了，不用重新跑所有任务。
边缘设备推理：比如在树莓派上跑小模型，设备不稳定，容易断。

5.2 优缺点

优点：
1. 不用改Ollama源码，实现简单，新手也能做。
2. 可以自定义检查点的保存间隔、保存路径，灵活度高。
3. 成本低：检查点文件很小（一个几十KB），不会占太多空间。
缺点：
1. 检查点是基于「已经生成的内容」续的，不是Ollama内部的完整推理状态（比如Ollama的注意力缓存），所以续的时候可能会有一点点偏差（比如内容重复、逻辑不一致），不过一般不影响使用。
2. 保存间隔如果设得太小（比如每10个token存一次），会增加磁盘读写的负担，影响推理速度。

5.3 注意事项

检查点文件的路径要固定，别随便改，不然下次找不到检查点。
不同的提示词要用不同的检查点，避免混了（上面的代码已经用提示词的前20个字符做了区分）。
如果推理的内容有特殊字符（比如中文、表情），存检查点的时候要指定编码为utf-8（上面的代码已经加了），不然会乱码。
不要在推理过程中修改Ollama的模型或者参数，不然续的时候可能会出错。

六、总结

这个检查点机制本质上就是给Ollama加了个「自动保存」功能，解决了长推理过程中意外中断的问题。实现起来很简单，不用复杂的技术，只要懂一点Python就能搞定。如果你的工作经常用到Ollama做长推理，这个机制能帮你省很多时间，避免之前的心血白费。

当Ollama意外退出后自动恢复推理进度的检查点机制实现思路

一、问题背景：为什么需要Ollama的自动恢复功能

二、核心实现思路：检查点机制的底层逻辑

2.1 核心原理拆解

三、具体实现步骤（附完整示例）

3.1 前置准备：先搭基础环境

3.2 完整实现代码（Python）

3.3 怎么测试这个脚本？

四、检查点机制的优化方向

4.1 自动清理旧检查点

4.2 加密检查点文件

4.3 定时检查Ollama状态

五、这个机制的应用场景、优缺点和注意事项

5.1 应用场景

5.2 优缺点

5.3 注意事项

六、总结

Comments

关联文章

AI助手

一、问题背景：为什么需要Ollama的自动恢复功能

二、核心实现思路：检查点机制的底层逻辑

2.1 核心原理拆解

三、具体实现步骤（附完整示例）

3.1 前置准备：先搭基础环境

3.2 完整实现代码（Python）

3.3 怎么测试这个脚本？

四、检查点机制的优化方向

4.1 自动清理旧检查点

4.2 加密检查点文件

4.3 定时检查Ollama状态

五、这个机制的应用场景、优缺点和注意事项

5.1 应用场景

5.2 优缺点

5.3 注意事项

六、总结

同分类面试题

Comments

关联文章

AI助手