OpenAI多轮对话历史消息管理：从messages到会话裁剪实践

最近在做基于OpenAI接口的智能问答机器人，功能本身不算复杂，但真正动手做多轮对话的时候，我发现了一个特别容易被忽略的问题：OpenAI的接口是无状态的，它根本不会记住你上一句说了什么。换句话说，每次调用API都得把之前的整段对话历史一起打包传过去，模型才能表现出“记得你刚才说过啥”的效果。官网上关于消息结构其实写得很清楚，但真正落地时怎么维护历史消息、什么时候裁剪、怎么裁剪、多用户并发怎么办，文档里并不会告诉你。这篇文章就当作我自己这次实现历史消息调用的一份实操记录，从最基础的messages结构讲起，到最后的会话管理器完整代码，以及在调试过程中踩过的一堆坑。希望能给正在做类似功能的同学一些参考。

1. 为什么必须自己维护历史消息

1.1 无状态接口带来的问题

先说一个很多人初学时的困惑：我明明在同一个客户端里连着问了两句话，为什么第二句话模型好像完全不知道我第一句说了什么？

原因很简单，OpenAI的Chat Completions接口在设计上就是无状态的。它接收一个messages数组，然后根据数组内容生成回复，处理完这一次请求之后，它不会在服务器端保存任何关于这段对话的“记忆”。第二次调用的时候，你传进去的如果是全新的对话数组，那模型对第一次的对话一无所知。

这就意味着“多轮对话”这件事，得由我们开发者自己来完成。你需要把历史上用户和助手产生的所有消息都记录下来，每次请求时把这份记录原封不动地附加到新的消息后面，一起提交给接口。模型看到完整的对话上下文之后，才能给出符合语境的回答。

我第一次写的时候就没想明白这一点，直接写了个单轮调用，用户在页面上连续提问，结果第二条回复完全跟第一条不搭边。后来才意识到，不是模型不聪明，是根本没人把历史对话给它看。搞清楚了这一点，后面的事情其实就顺理成章了。

1.2 历史对话的构成：三种角色

要维护历史消息，首先得理解messages数组里这个“消息”到底是什么结构。每个消息是一个对象，包含两个核心字段：

• role：消息的发送者角色
• content：消息的具体内容

在聊天场景下有三种角色：

• system：系统设定，用来给模型定义角色、行为规范。一般放在消息列表的第一条，且只放一条。
• user：用户的输入内容。多轮对话里就是每一次用户提问。
• assistant：助手（模型）的回复内容。每轮模型回答完，你要把它存下来，下次作为历史传回去。

举个例子，我有个理财咨询机器人，system内容可能是“你是一名资深的理财顾问，回答要简洁专业”。接下来每一轮用户提问就是user消息，模型回复就是assistant消息。这个列表会越来越长，每一轮对话都会往里面追加两条消息。

这里有一个容易忽略的细节：assistant消息是必须保存的，你不能只存user消息。因为模型生成回复时会参考整个对话的互动逻辑，如果只有用户的问题，没有之前的助手回复，上下文就是残缺的，模型很容易出现前后矛盾。后面我会专门讲这个坑。

2. 环境准备与基础调用

2.1 安装openai库，版本要选对

先说环境。我用的是Python 3.10，openai官方Python库。安装命令特别简单：

pip install openai

但是版本问题一定要留意。openai老版本是0.28.x，新版本是1.x系列，两者的调用方式完全不一样。0.x时代的写法是：

import openai openai.api_key = "sk-xxx" response = openai.ChatCompletion.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "你好"}] )

1.x之后废掉了这种全局配置的写法，改成了实例化客户端的方式：

from openai import OpenAI client = OpenAI(api_key="sk-xxx") response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "你好"}] )

区别在于，1.x版本把配置收敛到了OpenAI对象上，也更符合现代SDK的习惯。如果你在网上搜教程，发现代码对不上，大概率就是新旧版本混用了。我自己安装的时候没注意，结果用旧语法调了一个多小时才反应过来是版本问题。建议直接装最新版，用新的客户端写法。

至于API Key，到OpenAI平台的API Keys页面自己创建一个，然后设置成环境变量。代码里我建议用环境变量读取，而不是硬编码在源码里，避免密钥泄露：

import os client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

在终端里启动项目之前设置好环境变量，或者用.env文件管理。Windows用户别忘记用set OPENAI_API_KEY=sk-xxx，macOS/Linux用户用export OPENAI_API_KEY=sk-xxx。

2.2 跑通第一轮对话

环境没问题之后，先写个最简单的调用验证一下连通性。下面这个代码请求模型回答一句话，然后打印回复内容：

from openai import OpenAI client = OpenAI() response = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "你是一个热心的助手。"}, {"role": "user", "content": "请用一句话介绍一下历史消息调用的概念。"} ] ) print(response.choices[0].message.content)

这里的messages数组就是最基础的对话结构：先给系统设定角色，再放用户的问题。模型返回的是一个ChatCompletion对象，我们需要从response.choices[0].message.content中取出真正的文本回复。

第一次跑通的时候其实没啥成就感，因为这就是最基础的调用。但注意，这个数组就是后续所有多轮对话的“地基”。后续要做历史消息调用，本质就是往这个数组里持续追加user和assistant消息，而不是每次只传一个问题。

2.3 响应对象里还有什么信息值得看

很多人拿到回复之后只盯着choices[0].message.content看，其实response对象里有个我后来才发现很有用的字段：usage。它记录了本次请求消耗的token数量，结构大概是：

• prompt_tokens：请求里所有消息占用的token数
• completion_tokens：模型生成回复占用的token数
• total_tokens：两者之和

每次对话打印一下这个信息，对后面做历史裁剪非常有帮助。因为上下文窗口是有限的，到底还能塞多少消息，全靠token数量来判断：

print(response.usage.prompt_tokens, response.usage.completion_tokens, response.usage.total_tokens)

另外一个字段是response.choices[0].message.role，正常来说它的值应该是“assistant”。这在我们保存历史消息时可以用上，确保角色的标记和内容对上。

3. 历史消息调用的核心实现

3.1 最朴素方案：维护一个消息列表

理解了基础结构，历史消息调用最直接的做法就是用一个Python列表保存所有消息，每轮对话往里面追加内容，然后整体传给接口。下面这段代码是我早期验证时写的：

from openai import OpenAI client = OpenAI() messages = [] while True: user_input = input("你：") if user_input.lower() == "quit": break messages.append({"role": "user", "content": user_input}) response = client.chat.completions.create( model="gpt-4o-mini", messages=messages ) assistant_reply = response.choices[0].message.content messages.append({"role": "assistant", "content": assistant_reply}) print("助手：", assistant_reply)

逻辑很简单：用户输入内容，追加一条user消息；请求返回后，把助理回复追加一条assistant消息。下一次循环用户再输入时，messages里已经包含了之前所有对话，模型就能顺着上文继续回答。

实测效果也确实如此。第一轮问“我叫小明”，第二轮问“我叫什么”，模型能正确答出“小明”。这就是历史消息调用的最小实现。

但实际开发里不可能用全局列表，因为一个机器人可能要同时服务成百上千个用户，每个用户的对话上下文必须互相隔离。而且这个列表会无限增长，塞满上下文窗口只是时间问题。所以真正的落地代码需要考虑三件事：会话隔离、消息持久化、上下文裁剪。

3.2 会话管理器：让每个用户都有自己的上下文

我的做法是把消息列表封装到一个类里，每个会话实例持有自己独立的messages列表。这样不同用户之间的对话历史完全隔离，互不干扰。

类设计大概长这样：

import json from openai import OpenAI class ChatSession: def __init__(self, model="gpt-4o-mini", system_prompt="你是一个乐于助人的助手。", max_messages=10): self.client = OpenAI() self.model = model self.system_prompt = system_prompt self.max_messages = max_messages self.messages = [{"role": "system", "content": system_prompt}] def add_message(self, role, content): self.messages.append({"role": role, "content": content}) def trim_messages(self): # 保留system消息，限制user+assistant消息的总条数 while len(self.messages) > self.max_messages * 2 + 1: self.messages.pop(1) # 弹出最旧的一条，从system后面开始删 def chat(self, user_input): self.add_message("user", user_input) self.trim_messages() response = self.client.chat.completions.create( model=self.model, messages=self.messages, temperature=0.7 ) reply = response.choices[0].message.content self.add_message("assistant", reply) return reply def save(self, path): with open(path, "w", encoding="utf-8") as f: json.dump(self.messages, f, ensure_ascii=False, indent=2) def load(self, path): with open(path, "r", encoding="utf-8") as f: self.messages = json.load(f)

值得解释几个设计点。

第一，为什么裁剪消息时用pop(1)而不用pop(0)？因为messages列表的第0个元素是system消息，这个设定要一直保留在最开头，不能删掉。所以要删老消息时，从索引1开始删，也就是最早的user或assistant消息。

第二，为什么max_messages要乘以2再加1？因为每一轮对话会产生两条消息（1条user加1条assistant）。如果我想给模型保留最近10轮对话，那就要保存20条消息，再加上开头的1条system，总共就是21条。列表长度一旦超过这个值，就删掉最早的一轮。

第三，save和load方法用来做会话持久化。服务重启的时候，直接把之前保存的JSON文件读回来，就能恢复之前的对话状态。这在实际项目里很重要，否则服务一重启所有上下文全丢，用户体验会很差。

使用的时候极其简单：

session = ChatSession(system_prompt="你是一名耐心的英语老师。") print(session.chat("我英语基础很差，怎么开始学？")) print(session.chat("那每天学习多长时间合适？"))

第二个问题模型会结合前面聊的内容给出更贴合的回答，这就是历史消息调用的效果。

3.3 上下文窗口限制与裁剪策略

前面按条数裁剪的方案虽然简单，但有个问题：不同消息的长度天差地别。有人一条消息就几千字，有人几十个字。按条数裁剪，输在特别长的情况下照样会撑爆上下文窗口；输在特别短的情况下，又浪费了模型本来可以记住更多对话的能力。

所以更合理的做法是按token数量来裁剪。OpenAI官方提供了一个辅助库tiktoken，专门用来把文本切分成token：

pip install tiktoken

用起来也很直接：

import tiktoken def count_tokens(text: str) -> int: encoding = tiktoken.encoding_for_model("gpt-4o-mini") return len(encoding.encode(text))

然后可以重写trim逻辑，从最旧的消息开始删，直到总token数低于设定阈值：

def trim_by_tokens(self, max_tokens): # 计算所有消息外的system部分的token数 while True: total_tokens = sum( count_tokens(m["content"]) for m in self.messages ) if total_tokens <= max_tokens or len(self.messages) <= 2: break removed = self.messages.pop(1) print(f"丢弃了一条旧消息：{removed['role']}，节省了 {count_tokens(removed['content'])} tokens")

这里需要注意，我是按纯文本内容估算token数，实际API计算时会加上角色标记、消息格式等额外开销。不过作为裁剪依据已经足够。真正严谨的做法是把整个messages数组用ChatML格式编码再统计，但对大多数场景来说，内容token估算加一个安全余量就够了。

用token裁剪时还要考虑“输出token的预留空间”。假设上下文窗口是128k，我通常会把历史消息的token上限设置为100k，给模型生成回复留出28k的余量。如果你把历史消息塞到127k，模型可能连一句回复的空间都没有，直接报错。这是我实测中踩过的坑，后面“问题排查”会细说。

那到底留多少合适？我个人的经验是：输出token设置成max_tokens后，历史token阈值不要超过“上下文总长度减max_tokens再减一个缓冲”。比如max_tokens设为4096，总长度128k，那历史阈值就是128k减4k减2k缓冲，大概122k以内。不同模型的上下文长度不一样，去模型对应的官方文档页面看一眼数值再计算。

3.4 多用户并发怎么处理

聊天机器人部署之后肯定不是一个人用，每个用户都得有独立的会话状态。最简单的做法是用一个字典暂存所有会话，以用户ID作为key。

sessions = {} def get_session(user_id: str) -> ChatSession: if user_id not in sessions: sessions[user_id] = ChatSession() return sessions[user_id] def handle_message(user_id: str, message: str) -> str: session = get_session(user_id) return session.chat(message)

这种方式适合单进程、会话量不大的情况。一旦用户量变多，内存压力会增大，服务重启会话也会丢，所以在生产环境我会建议把会话数据放到Redis这类外部存储里，会话消息用JSON序列化后按key存储，读的时候反序列化回messages数组。

这里还有个并发安全的小细节：如果同一个用户的请求是并发进来的，两条请求同时往同一个messages列表里追加消息，会导致顺序错乱和上下文污染。所以在封装会话的时候，我额外加了个threading.Lock，保证同一会话的读写是串行的：

import threading class ChatSession: def __init__(self, ...): ... self.lock = threading.Lock() def chat(self, user_input): with self.lock: # 原有的追加、调用、返回逻辑 ...

别小看这个锁。我一开始没加，压测的时候发现有的回复串了上下文，用户A的提问被用户A同会话的另一条并发请求抢占了顺序，回复内容看起来就很奇怪。加上锁之后这个问题再没出现过。

4. 常见问题与排查技巧实录

4.1 报错速查表

这段内容集中整理一下，方便以后再看。

最常见的还是上下文超长那个报错。错误信息会给出一大段提示，核心一看就能明白：这段对话已经突破模型的上限了。看到这个错误别慌，思路就一个：把最老的对话历史删掉，或者删掉一部分system提示词，直到总数降下来。如果是我的会话管理器，直接调用裁剪函数就行。

4.2 角色顺序与消息丢失问题

除了超长，我们在调试历史消息时最容易犯的错误有三个。

第一个是只存user不存assistant。后果是模型每次回复都像在“自言自语”第一句，完全不知道它之前说过什么。检查方法很直接：打印messages数组，看看有没有规律的“user、assistant、user、assistant”交替。如果出现两个user连续挨着，中间没有assistant，说明上一轮的回复没存进去。

第二个是忘了system消息最开头且只有一条。按官方规范，system消息是可选的，但一旦存在，就必须放在列表第一位。我见过有人中途又在列表里插入第二条system消息，模型会乱掉，表现行为异常。

第三个是消息裁剪时误删了system。我之前写过一版裁剪，直接从pop(0)开始删，结果把system删了，模型立刻失去人设，回答风格完全变了。修这个bug我们已经在裁剪函数里做过约束，从索引1开始删，这里再提醒一次：system消息就是地基，永远不能动。

4.3 怎么确认模型真的“记住”了历史

排查历史消息是否生效，其实有个很简单的测试方法。

跑完一轮对话后，第二轮问一个只有结合第一轮才能回答的问题。比如第一轮让模型给自己起个名字叫“小光”，第二轮问“你叫什么名字”，如果模型答出“小光”，说明历史消息生效了；如果模型说“我没有名字”，说明消息没传进去或者没传完整。

我还会在每次请求前打印一个摘要信息，帮助定位问题：

def chat(self, user_input): ... print(f"请求{self.model}，当前消息数量：{len(self.messages)}，" f"消息token总数：{count_tokens(str(self.messages))}")

而且我习惯把response.usage的两个值也输出出来，观察每次请求的prompt_tokens是否在逐步增加。如果它一直在涨，说明历史确实在累积；如果一直不变，说明你可能每次调用的都是同一个被重置的列表，那问题多半出在对象的复用上——很常见的错误是每次请求都新建了一个ChatSession或者重新初始化了messages。

4.4 一个印象深刻的内存泄漏坑

这里多聊一句我在做会话管理时遇到的一个比较隐蔽的问题。一开始我用一个简单的dict保存所有用户会话，没有上限。结果线上跑了两天后，内存占用直线飙升。排查后发现，有些用户创建了会话之后就不再活跃了，但他们的messages列表依然留在内存里，白白占着资源。

后来我加了两层保护：一层是设置最大会话数，超出就淘汰最久没活动的会话；另一层是给每个会话记录最后活跃时间，超过一定时限就自动清理。内存问题这才解决。如果你做的是生产环境服务，这个点值得提前考虑到。

5. 进阶优化：让历史消息方案更省、更快

5.1 长对话的摘要压缩策略

前面说的裁剪策略是直接丢弃最早的对话，简单粗暴，但副作用是模型会丢失比较久远的关键信息。比如用户在第2轮说过“我是做IT行业的”，而对话已经持续了40轮，这段信息被裁剪了，后面再问职业相关问题时模型就答不上来了。

更好的方案是“摘要压缩”：当历史消息快要触及token上限时，不直接丢弃旧消息，而是把旧消息交给模型，让它提炼成一段精简摘要，剩下的空间只保留最近几轮完整消息。下次请求时，messages变成这样：

• system：原始系统提示词
• user：一段摘要，比如“用户是一名IT行业从业者，正在咨询转行做产品经理的可能性，目前已经讨论了职业规划和时间安排”
• 后续：近期几条完整对话

这样做有两个好处：保留长期记忆，同时也控制了token总量。代价是要多一次模型调用。我在项目里是等对话长度达到阈值时才触发一次摘要，平时不调用，所以成本增加很有限。

5.2 在输出侧控制token成本

很多人在调用时忽略max_tokens参数，结果模型有时候输出长长一篇没人看的内容，成本高还慢。给文本类回复设置一个合理的最大输出长度很重要。比如我的问答机器人，max_tokens设置为1024，大多数回复足够用，还不会失控。

另外是模型选择。同样的上下文，gpt-4o-mini比gpt-4o便宜很多。我的策略是：日常问答用gpt-4o-mini，只有需要复杂推理或长文分析时才切换到gpt-4o。同一个会话管理器里，只要把model参数抽出来，随时可以切换。

5.3 会话持久化的轻量方案

如果是单机小项目，用JSON文件保存会话最省事，我在会话管理器里已经写了save和load方法。文件命名直接用会话ID，比如session_10001.json，每小时自动落盘一次。服务重启后扫描目录，再把所有会话加载回内存。

如果是多实例部署或者会话量很大，那就得上Redis了。Redis的好处是天然支持过期时间，可以给每个会话设置TTL，活跃会话一直续期，不活跃的自动过期清理。存储结构也很简单，key是用户ID，value是messages数组的JSON字符串。

import redis import json r = redis.Redis(host="localhost", port=6379, db=0) def save_session(user_id: str, messages: list): r.set(f"session:{user_id}", json.dumps(messages, ensure_ascii=False), ex=3600) def load_session(user_id: str): data = r.get(f"session:{user_id}") if data: return json.loads(data) return None

Redis的方案还有个额外好处，不同后端语言都能直接读写这套数据结构，以后如果要把机器人接入别的服务，数据这块不用重新设计。

6. 最后想说的几点体会

用OpenAI库实现历史消息调用，看起来只是维护一个数组，但真正做下来会发现它牵涉到会话生命周期管理、token预算控制、并发隔离、持久化设计，每一环都需要认真考虑。

我个人在操作中最大的体会是：不要把历史消息功能当成一个“辅助功能”来看，它本身就是一个完整的子系统。你越早把它模块化，后面接持久化、接并发、接摘要压缩就越轻松。我最初把所有逻辑塞进一个业务函数里，后来改造成独立的ChatSession类，代码反而短了很多，因为职责清晰了。

最后再分享一个小技巧：调试这类功能时，一定要写一个简单的自动化测试脚本，模拟连续多轮的对话，每一轮都打印当前messages数组和usage信息，然后把输出保存为日志。这样每次改动后跑一遍，就能快速发现“角色没交替”“上下文被清空”“内存悄悄涨”这类隐性问题。

如果你现在正准备给自己的聊天机器人加上多轮记忆能力，建议先按文章里第三部分的会话管理器把骨架搭起来，跑通用户隔离和历史累积，再逐步加裁剪和持久化。基础打稳了，后面加什么功能都不慌。