资讯详情

LLM 多轮对话状态管理:从无状态 API 到有状态会话的 TaoToken 实践

📅 2026/10/3 12:19:17 | 华诺云谱 👁 阅读
LLM 多轮对话状态管理:从无状态 API 到有状态会话的 TaoToken 实践
1. 无状态 API 为什么总在第三轮对话开始“失忆”如果你用最朴素的方式调过大模型 Chat API大概率经历过这个场景第一轮问“帮我写个 Python 读取 CSV 的函数”模型答得挺好第二轮说“改成支持分块读取”它却反问你“什么 CSV要读什么”——不是模型笨而是你根本没把上一轮的消息带过去。大模型的 Chat Completions 接口本质上是无状态的。服务端不记得你上一秒说过什么每次 HTTP 请求都是一个独立的宇宙。你想让它“记得”唯一的办法就是在这一次请求的messages数组里把之前所有的 user / assistant 消息按顺序重新塞一遍。这就是 LLM 多轮对话状态管理最底层的约束状态不在服务端而在你的客户端代码里。这件事听起来简单真做起来会撞上三堵墙。第一堵墙是上下文窗口有限。主流模型的上下文从 8K 到 128K token 不等看着挺大但一轮对话动辄几百上千 token二十轮下来很容易顶到天花板。一旦超出要么请求直接报错要么模型开始“遗忘”最早的内容。第二堵墙是Token 成本随轮次线性增长。假设每轮问答平均 500 token第 1 轮你发 500第 10 轮你要发 5000第 20 轮你要发 10000。用户聊得越久你越烧钱而且烧的钱大部分花在重复发送历史消息上。第三堵墙是状态漂移。就算你老老实实把历史全带上模型也可能在长上下文里抓错重点。用户第三轮说的“就按刚才那个格式”到第十轮模型已经分不清“刚才”指的是哪一版格式了。信息没丢但语义焦点丢了。所以 LLM 多轮对话状态管理要解决的核心问题不是“怎么存历史”而是在有限的上下文窗口里保留对当前这轮对话最有价值的信息同时把 Token 消耗压住。围绕这个目标业界通常拆成四个机制来做消息压缩、摘要替换、关键信息提取、会话持久化。下面我会用 TaoToken 作为统一的 API 通道把这套东西从零搭一遍你能直接复制去跑。先说清楚 TaoToken 在这里扮演什么角色。它是一个统一的大模型 API 网关你用一个 Key 就能访问多家模型Base URL 固定省得每换一个模型就改一遍 SDK 配置。对多轮对话场景来说这点很关键——因为状态管理逻辑里经常要调用“摘要模型”和“主对话模型”如果两个模型来自不同厂商、走不同通道你的代码里会塞满各种 base_url 判断。统一通道之后切换模型只是改一个 model 字符串的事。适合读这篇的人正在做客服机器人、销售助手、代码助手这类需要连续多轮交互的后端同学或者你已经用单轮 API 跑通了 demo但一上多轮就发现上下文乱套、成本失控。接下来从环境准备到可复制的会话管理器再到真实报错排查一步步来。2. TaoToken 前置准备统一 Key 与 API 通道配置在写会话管理器之前得先把调用通道打通。这一步不做后面所有代码都跑不起来。我用 TaoToken 的原因是它把多模型的接入收敛成一个 Base URL 加一个 Key配置一次到处能用特别适合我们这种要在摘要模型和主模型之间来回切的场景。2.1 获取 API Key 与确认 Base URL先到控制台创建一个 API Key。地址是https://taotoken.net/console登录后在 API Keys 页面点创建复制出来的那串sk-开头的字符串就是你的凭证。注意它只完整显示一次先存到环境变量里别硬编码进代码。Base URL 统一用https://taotoken.net/api注意这里不带任何查询参数。很多人第一次配会把官网地址和 API 地址搞混官网是https://taotoken.net/但 SDK 里填的 base_url 必须是带/api的那个否则请求会打到网页服务器上返回 HTML你的 JSON 解析直接崩。我习惯用环境变量管理Linux / macOS 下这样写export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/apiWindows PowerShell 用$env:TAOTOKEN_API_KEYsk-你的key $env:TAOTOKEN_BASE_URLhttps://taotoken.net/api2.2 用 OpenAI SDK 验证通道TaoToken 的接口兼容 OpenAI 的 Chat Completions 协议所以直接用 openai 官方 SDK 就行不用装额外的包。先装依赖pip install openai然后写一个最小验证脚本确认 Key 和 Base URL 都对import os from openai import OpenAI client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 只回复两个字通了}], ) print(resp.choices[0].message.content)跑出来打印“通了”说明通道没问题。如果报 401八成是 Key 复制时带了空格或者引号如果报连接错误检查 base_url 是不是漏了/api。2.3 为什么多轮对话特别需要统一通道这里多说一句设计上的考虑。多轮对话状态管理里我们至少要用到两类模型调用一类是主对话模型负责和用户交互另一类是摘要模型负责把早期历史压缩成短文本。这两类调用对模型的要求不一样——主模型要强摘要模型要便宜快。如果它们走不同的厂商通道你的代码里就得维护两套 client、两套鉴权、两套错误处理。用 TaoToken 之后两个调用共用一个 client只是model参数不同# 主对话 main_resp client.chat.completions.create(modelgpt-4o, messagescontext) # 摘要压缩 sum_resp client.chat.completions.create(modelgpt-4o-mini, messagessummary_prompt)切换模型、对比效果、做 A/B 测试都只是改一个字符串。对状态管理这种需要反复调优压缩策略的场景省下来的配置时间相当可观。通道打通后下面进入正题会话管理器怎么写。3. 可复制的会话状态管理器配置与实现这一节是全文的技术核心。我会给出一个能直接跑的 Python 版本会话管理器包含会话存储、上下文窗口管理、摘要压缩、实体提取四个部分。代码结构参考了生产环境的做法但做了简化你可以按需替换存储后端。3.1 会话状态的数据结构先定义两个基础结构单条消息和整个会话状态。消息除了 role 和 content我还带了 token 估算值和时间戳方便后面做窗口裁剪。from dataclasses import dataclass, field from typing import List, Dict import time dataclass class ChatMessage: role: str # system / user / assistant content: str timestamp: float field(default_factorytime.time) token_count: int 0 dataclass class ConversationState: session_id: str history: List[ChatMessage] field(default_factorylist) entities: Dict[str, str] field(default_factorydict) total_tokens: int 0 def add_message(self, msg: ChatMessage): self.history.append(msg) self.total_tokens msg.token_count def history_tokens(self) - int: return sum(m.token_count for m in self.history)token 估算这里先用一个粗糙的公式中文字符约 1 字 1 token英文约 4 字符 1 token。生产环境建议换成 tiktoken 之类的精确分词器但做逻辑验证够用了。def estimate_tokens(text: str) - int: # 粗略估算中文按字符数英文按 4 字符 1 token chinese sum(1 for c in text if \u4e00 c \u9fff) other len(text) - chinese return chinese max(1, other // 4)3.2 上下文窗口管理的三段式策略这是整个管理器最关键的部分。当历史 token 超出预算时不能简单粗暴地砍掉最早的几条那样会丢失关键约束。我采用的是“系统消息 摘要 最近消息 实体信息”的四段式组装其中摘要和实体信息负责保留远期记忆最近消息负责保留即时上下文。class ContextWindowManager: def __init__(self, max_tokens: int 8000): self.max_tokens max_tokens def build_context(self, state: ConversationState, client) - List[dict]: history state.history if state.history_tokens() self.max_tokens: return [{role: m.role, content: m.content} for m in history] # 预算分配最近消息 60%摘要 30%实体 10% budget self.max_tokens system_msgs [m for m in history if m.role system] for m in system_msgs: budget - m.token_count recent_budget int(budget * 0.6) summary_budget int(budget * 0.3) recent self._take_recent(history, recent_budget) early history[: len(history) - len(recent)] context [] context.extend({role: m.role, content: m.content} for m in system_msgs) if early: summary self._summarize(early, summary_budget, client) context.append({role: system, content: f[对话摘要] {summary}}) if state.entities: ent_text 已知信息: ; .join(f{k}{v} for k, v in state.entities.items()) context.append({role: system, content: ent_text}) context.extend({role: m.role, content: m.content} for m in recent) return context def _take_recent(self, history, budget): recent, used [], 0 for m in reversed(history): if used m.token_count budget: break recent.insert(0, m) used m.token_count return recent def _summarize(self, messages, max_tokens, client) - str: text \n.join(f{m.role}: {m.content} for m in messages) prompt ( f把下面的对话压缩成不超过 {max_tokens} token 的摘要 f保留关键决策、约束条件和用户偏好去掉寒暄\n{text} ) resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], ) return resp.choices[0].message.content注意_summarize里用的是便宜的小模型因为摘要任务对模型能力要求不高用大模型纯属浪费。这也是统一通道的好处——同一个 client 换个 model 名就行。3.3 实体提取让关键信息不随历史被压缩摘要再聪明也会有信息损失所以对结构化的关键信息订单号、日期、用户偏好我单独抽出来存进entities每轮都注入上下文。这样即使原始消息被摘要替换这些信息依然在。import re def extract_entities(state: ConversationState, message: str): patterns { order_id: r订单号[:]\s*(\w), date: r(\d{4}[-/]\d{2}[-/]\d{2}), email: r([\w.-][\w-]\.[\w.]), } for key, pat in patterns.items(): m re.search(pat, message) if m: state.entities[key] m.group(1)规则提取覆盖高频模式够用且零延迟。等你有精力了再上 NER 模型处理“上周三那个单子”这种口语表达。3.4 会话持久化配置会话状态不能只放内存进程一重启就全没了。我用 Redis 存热会话设置 1 小时 TTL冷数据异步落库。配置片段如下import json import redis class SessionStore: def __init__(self, hostlocalhost, port6379, ttl3600): self.r redis.Redis(hosthost, portport, decode_responsesTrue) self.ttl ttl def save(self, state: ConversationState): key fconv:{state.session_id} data { history: [m.__dict__ for m in state.history], entities: state.entities, total_tokens: state.total_tokens, } self.r.setex(key, self.ttl, json.dumps(data, ensure_asciiFalse)) def load(self, session_id: str): raw self.r.get(fconv:{session_id}) if not raw: return None data json.loads(raw) state ConversationState(session_idsession_id) state.entities data[entities] state.total_tokens data[total_tokens] for m in data[history]: state.history.append(ChatMessage(**m)) return state如果你本地没装 Redis可以先用一个内存字典顶上接口保持一致后面换真存储不用改调用方代码。3.5 组装完整管理器把上面几块拼起来就是对外暴露的process_messageclass ConversationManager: def __init__(self, client, store, max_tokens8000): self.client client self.store store self.window ContextWindowManager(max_tokens) def process_message(self, session_id: str, user_input: str) - str: state self.store.load(session_id) or ConversationState(session_idsession_id) user_msg ChatMessage(user, user_input, token_countestimate_tokens(user_input)) state.add_message(user_msg) extract_entities(state, user_input) context self.window.build_context(state, self.client) resp self.client.chat.completions.create( modelgpt-4o, messagescontext, ) reply resp.choices[0].message.content reply_msg ChatMessage(assistant, reply, token_countestimate_tokens(reply)) state.add_message(reply_msg) self.store.save(state) return reply到这里一个具备上下文压缩、实体保留、持久化能力的会话管理器就成型了。下一节验证它到底记不记得住。4. 多轮对话验证从请求到上下文一致性检查代码写完不验证等于没写。这一节我用一个跨多轮的测试脚本检查三件事短期记忆是否保持、远期信息是否在压缩后仍可召回、Token 消耗是否被压住。4.1 构造一个会“埋信息”的测试对话测试思路是第一轮埋一个关键信息比如订单号中间插入若干轮无关闲聊把历史撑长最后再问那个订单号。如果管理器工作正常模型应该能答出来。manager ConversationManager(client, SessionStore()) sid test-session-001 # 第 1 轮埋订单号 print(manager.process_message(sid, 我的订单号是 A12345帮我查下状态)) # 第 2-8 轮无关闲聊撑大历史 for i in range(2, 9): manager.process_message(sid, f随便聊点别的第 {i} 个话题今天天气不错) # 第 9 轮召回测试 print(manager.process_message(sid, 我刚才说的订单号是多少))4.2 观察成功结果正常输出应该类似第 1 轮回复好的订单 A12345 的状态是... ... 第 9 轮回复您刚才提到的订单号是 A12345。关键看第 9 轮。如果模型答出 A12345说明实体提取生效了——即使中间 7 轮闲聊把原始消息挤进了摘要区entities里的订单号依然被注入到了上下文。如果答不出来说明实体提取的正则没匹配上或者注入逻辑没生效。4.3 检查 Token 消耗曲线再验证一下压缩是否真的省了钱。在process_message里加一行日志打印每轮实际发送的 token 数context self.window.build_context(state, self.client) sent_tokens sum(estimate_tokens(m[content]) for m in context) print(f[session{session_id}] 本轮发送 {sent_tokens} tokens, 历史共 {state.history_tokens()} tokens)跑完 9 轮你会看到前几轮发送量随历史增长但一旦历史超过max_tokens发送量就被压在一个稳定值附近不再上涨。这就是三段式窗口管理的效果——历史继续变长但发给模型的上下文被摘要和裁剪控制住了。4.4 用统一通道做模型对比因为走的是 TaoToken 统一通道你可以把主模型从gpt-4o换成别的跑同一套测试脚本对比不同模型在长上下文里的召回能力。改一行就行resp self.client.chat.completions.create(modelclaude-3-5-sonnet, messagescontext)这种对比在排查“到底是状态管理的问题还是模型能力的问题”时特别有用。如果换个模型召回就正常了那说明你的压缩策略没问题是原模型长上下文能力弱。5. 常见报错排查401、local proxy failed 与 choices 解析失败多轮对话场景的报错比单轮更隐蔽因为错误可能出在摘要调用、主调用、存储三个环节中的任意一个。下面是我实际踩过的几类按报错原文对照排查。5.1 401 Unauthorized报错原文通常是openai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}排查顺序第一确认环境变量TAOTOKEN_API_KEY真的被读到了在脚本里print(os.environ.get(TAOTOKEN_API_KEY)[:8])看前几位第二检查 Key 有没有多余空格或换行从控制台复制时容易带上第三确认 Key 没有过期或被删除。多轮场景里如果摘要调用和主调用用了不同的 client 实例可能出现一个配对了、一个没配对的情况统一用一个 client 能避免这类问题。5.2 local proxy failed / connection error报错原文类似openai.APIConnectionError: Connection error.或者带local proxy failed字样。这类基本是网络层问题不是 Key 的问题。排查确认base_url是https://taotoken.net/api别写成官网地址确认本机没有残留的代理环境变量干扰echo $HTTP_PROXY看一下如果有就临时 unset 掉再试。多轮场景里如果摘要调用超时整个process_message会抛异常建议给摘要调用单独包一层 try失败时降级为直接截断而不是让整轮对话崩掉。5.3 reading choices 解析失败报错原文TypeError: NoneType object is not subscriptable或者KeyError: choices。这通常发生在你直接对响应做resp[choices]而不是resp.choices或者响应体根本不是预期的 JSON。最常见的原因是 base_url 配错请求打到了网页服务器返回的是 HTMLSDK 解析失败。另一个原因是模型名写错了某些通道对未知模型会返回非标准错误体。检查model字符串拼写以及 base_url 是否带/api。5.4 上下文超限报错报错原文This models maximum context length is 8192 tokens, however you requested 9500 tokens这说明你的窗口管理没生效或者max_tokens设得比模型实际窗口还大。检查两点ContextWindowManager的max_tokens要留出回复的余量比如模型窗口 8192你设 6000 比较稳确认build_context真的被调用了而不是某条分支直接返回了完整历史。5.5 会话状态丢失现象是重启服务后之前聊的内容全没了。这基本是存储层的问题确认 Redis 真的在跑redis-cli ping返回 PONG确认save在每轮结束后被调用确认 TTL 没设得太短。如果用的是内存字典版本那重启丢失是预期行为换 Redis 即可。5.6 三件套配置速查如果你用的是 Cline、CC Switch 这类工具接入配置项永远是三件套缺一不可配置项值Base URLhttps://taotoken.net/apiAPI Key控制台创建的sk-开头字符串Model ID如gpt-4o、claude-3-5-sonnet等任何“连不上”的问题先回头核对这三项八成能定位。6. 把状态管理接进你的真实项目走到这里你已经有了一个能跑的多轮对话状态管理器。最后说几个把它接进真实项目时的实用建议都是我在实际项目里踩出来的。第一摘要策略要按业务调。客服场景里“约束条件”最重要摘要 prompt 里要强调保留闲聊场景里“最近几轮”最重要可以把最近消息的预算从 60% 提到 80%。别一套参数用到底。第二实体提取先规则后模型。正则能覆盖 80% 的高频模式剩下 20% 的口语表达再考虑上 NER。一上来就上模型延迟和成本都不划算。第三给摘要调用加降级。摘要模型偶尔会超时或返回空这时候不要让整轮对话失败直接退化成“截断最早的消息”也能用。健壮性比完美压缩重要。第四监控发送 token 数。在process_message里埋点把每轮实际发送的 token 数上报。一旦发现某类会话的发送量持续上涨说明压缩策略对这类对话失效了需要针对性调整。如果你还没配好通道先去https://taotoken.net/api-keys拿 Key接入细节看https://taotoken.net/doc。想先直观感受一下多轮对话的效果可以直接在https://taotoken.net/models里对话测试把同一段对话复制进去观察模型在不同上下文长度下的表现差异。长期要做编码助手或 Agent 这类高频多轮场景的https://taotoken.net/coding-plan里的方案对控制成本更有帮助。状态管理这件事代码写对只是第一步参数调对才是长期功夫。
📝

华诺云谱内容团队

资深建站顾问 · 行业研究员

10年+企业数字化服务经验,专注智能建站、SEO优化与品牌营销,持续输出建站技巧、行业洞察与营销干货,已帮助5000+企业实现数字化增长。

你可能需要的服务

订阅华诺云谱资讯周报

每周一封,精选建站技巧、SEO与营销干货,直达邮箱。已有 8,000+ 企业主订阅,助你少走弯路。

↑