Python+ollama开发长时记忆对话

sht原创2025/8/3...大约 4 分钟

1. 开发计划

✅ V0.1：基础对话系统（本地调用 Ollama）

基础的 Python 调用 Ollama 对话代码

使用 Python 实现与 Ollama 的基础对话接口；
用户输入 → LLM 响应流程；
终端交互版本（命令行）；
「待定」抽象封装一个 ChatModel 接口类，便于后续接入其他模型。

🧠 V0.2：增加短时记忆能力（对话上下文）

拥有基础的短时记忆能力；

将历史对话内容拼接传入，模拟短时记忆；
上下文缓存机制（如 recent_n_turns）；
支持用户手动清除上下文（/reset 命令）。

💾 V0.3：数据持久化（本地 + 云）

对话数据持久化存储「本地化存储 and 云存储」

📁 本地化存储
- JSON / SQLite 存储每轮对话（按会话分组）——本地化数据存储格式；
- 提供导入/导出对话记录的功能；
- 本地文件结构设计示意：
```
/data/
  session_2025_05_25_14_30.json
```
☁️ 云端存储
- 设计简单 API 接口结构（RESTful）；
- 支持通过 requests 同步数据至远程数据库（如 LeanCloud、Firestore、Supabase）；
- 云端数据加密传输与权限验证（Token 或 API Key）。
- 云存储数据存储格式，以及数据如何进行交互；

🧬 V0.4：长时记忆与总结压缩

Ollama 对用户历史对话数据进行总结记忆；
对话数据过长，可以考虑分块（分批次）总结并存储——实现数据压缩，防止超出 token 长度；

定期对对话进行总结（e.g. 每 20 轮对话压缩一次）；
使用 LLM 总结模块，按分块进行摘要生成；
存储压缩版本作为“记忆摘要”，减少上下文 Token 压力；
构建一个 MemoryManager 组件，负责记忆管理、压缩、合并等操作。

🌐 V0.5：模型动态切换（本地 vs 在线）

Gemini 在线大模型选用——封装一个函数，在程序运行过程中，让用户选择所需要使用的大模型「Local or Server」；
注意点⚠️：Ollama 与 Gemini 的对话结构不同，需要做出适配；

实现模型选择接口：use_model("ollama") 或 use_model("gemini")；
抽象不同模型输入输出结构；
- Ollama: { 'role': 'user', 'content': '...' }
- Gemini: { 'parts': ['...'] }
自动适配输入格式、输出格式；
提供配置项或命令行参数选择模型来源（支持热切换）；
检查网络连通性与 API Key 存在性做出提示。

v0.1

1. 基础代码

我们先使用 Ollama 提供的库，来进行实现基础调用：

from ollama import chat

messages = [
  {
    'role': 'user',
    'content': '你好',
  },
]

response = chat('deepseek-r1:7b', messages=messages)
print(response['message']['content'])

在实现基础的单次调用后，我们可以开始尝试开发更多功能。

2. （扩展）逐字输出

想要实现大模型的逐字输出回复功能，我们需要使用到 Ollama 给我提供的 stream=True 的参数功能，先进行代码测试试验：

from ollama import chat
userinput = input('用户输入：')
messages = [
  {
    'role': 'user',
    'content': userinput,
  },
]

response = chat('deepseek-r1:7b', messages=messages,stream=True)
for messages in response:
  print(messages['message']['content'],end='',flush=True)

flush用来保证输出流畅

3. 封装输出

from ollama import chat
userinput = input('用户输入：')
messages = [
  {
    'role': 'user',
    'content': userinput,
  },
]
response = chat('deepseek-r1:7b', messages=messages, stream=True)

def stream_print(response):
  for messages in response:
    print(messages['message']['content'], end='', flush=True)

stream_print(response=response)

4.实现思维链封装

• 功能一：用户可以选择是否显示思维链；
• 功能二：默认情况下，不显示思维链；

from ollama import chat
think_switch , userinput= int(input('是否开启思维链(是输入1，否输入0)：')) , input('用户输入：')
messages = [
  {
    'role': 'user',
    'content': userinput,
  },
]
response = chat('deepseek-r1:7b', messages=messages, stream=True)

think_count = 0
def stream_print(response):
  global think_count
  global think_switch
  for messages in response:
    if think_switch==0:
      if think_count == 2:
        print(messages['message']['content'], end='', flush=True)
      if messages['message']['content'] == '<think>' or messages['message']['content'] == '</think>':
        think_count += 1
    else:
      print(messages['message']['content'], end='', flush=True)

stream_print(response=response)

5. 实现大模型回复开头：“思考中…”效果（可选、额外）

加上 思考中…效果：

在真正开始回答前，先打印 ...，每个点之间有停顿，看起来像人在思考；
然后再进入逐字打字输出（带随机停顿和标点停顿）；

from ollama import chat
import time

think_switch , userinput= int(input('是否开启思维链(是输入1，否输入0)：')) , input('用户输入：')
messages = [
  {
    'role': 'user',
    'content': userinput,
  },
]
response = chat('deepseek-r1:7b', messages=messages, stream=True)


def thinking_effect(thinking_time=2.0, interval=0.5):
  """思考中... 效果"""
  print("思考中", end='', flush=True)
  start = time.time()
  while time.time() - start < thinking_time:
    print(".", end='', flush=True)
    time.sleep(interval)
  print("\n")


think_count = 0
def stream_print(response):
  global think_count
  global think_switch
  for messages in response:
    if think_switch==0:
      if think_count == 2:
        print(messages['message']['content'], end='', flush=True)
      if messages['message']['content'] == '<think>' or messages['message']['content'] == '</think>':
        think_count += 1


    else:
      print(messages['message']['content'], end='', flush=True)

thinking_effect()
stream_print(response=response)