企业级意图识别不是一个”分类器”，而是一条 6 层处理链路。用户输入从进入系统到最终路由，依次经过：用户原始输入
│
▼
① 输入预处理（Query Preprocessing） ← 把”脏输入”变成”干净输入”
│
▼
② 安全护栏（Guard Rails） ← 拦截恶意/违规/超范围请求
│
▼
③ 意图分类（Intent Classification） ← 判断用户想干什么 + 打置信度
│
▼
④ 实体提取 / 槽位填充（Entity Extraction） ← 提取业务关键参数
│
▼
⑤ 多意图拆分（Multi-Intent Resolution） ← 一句话多个意图时拆开处理
│
▼
⑥ 置信度决策路由（Confidence Routing） ← 高/中/低置信走不同路径
│
▼
路由到对应业务处理节点
实际项目中 ③④⑤ 通常合并为 一次 LLM 调用 + 结构化输出（JSON Mode / Tool Call），一次性返回意图、置信度、实体、子意图。下文按逻辑分层说明，但不代表一定是 6 次独立调用。
◆
一、输入预处理（Query Preprocessing）
1.1 要解决什么问题
用户的原始输入是”脏”的——有错别字、口语化表达、省略指代、中英混杂、多余空格符号等。如果直接拿去做分类，会严重影响准确率。预处理的目标是把”脏输入”标准化为”干净输入”。
1.2 具体做哪些事
| 处理项 | 说明 | 示例 | 常用工具/方案 |
|---|---|---|---|
| 文本清洗 | 去除首尾空白、合并连续空格、统一全角/半角标点 | " 你好！！ " → "你好!" | 正则表达式，Python str.strip() |
| 错别字纠正 | 修正常见业务领域错别字 | "帐号登陆" → "账号登录" | LLM 纠错 / pycorrector / 自建业务纠错词典 |
| 口语改写 | 将口语化、方言化表达转换为标准表达 | "这玩意儿咋整啊" → "这个产品怎么使用" | LLM query rewrite |
| 指代消解 | 将”它””那个””上面说的”替换为上下文中的具体实体 | "那个还有吗" → "iPhone 16 还有库存吗" | LLM（传入对话历史） |
| 多轮上下文合并 | 将当前轮和历史对话合并成完整语义 | 历史："查一下我的订单" 当前："退了吧" → "我要退掉我的订单" | LLM（拼接 history + current） |
| 语言检测 | 识别输入语言，决定后续用哪套 prompt/模型 | 中文/英文/日文 | langdetect / LLM |
1.3 实际项目怎么做
•小规模项目：用正则 + 业务词典处理清洗和纠错，不调 LLM，速度快成本低•中大规模项目：用一次轻量 LLM 调用做 query rewrite，prompt 示例：你是一个查询改写助手。将用户的口语化输入改写为清晰、完整、标准的表达。
要求：
1. 修正错别字
2. 将口语/方言转为书面表达
3. 根据对话历史补全指代和省略
4. 不要改变用户的原始意图
5. 只输出改写后的文本，不要解释
对话历史：{history}
用户当前输入：{current_input}
改写后：
1.4 注意事项
•预处理不能改变用户意图，只做”等价改写”•纠错要建业务词典而不是通用纠错（通用纠错可能把专业术语”纠”错）•多轮上下文合并是高价值环节——很多意图识别不准的根因是缺少上下文◆
二、对话历史管理（预处理的核心前置依赖）
上面的指代消解、省略补全都依赖对话历史。但 LLM 本身没有记忆——每次调用都是无状态的。所谓”拿到历史”，就是 你把历史存下来、拼进 prompt、传给它。用户第 1 轮：”查一下我的订单”
用户第 2 轮：”退了吧” ← 模型如果看不到第 1 轮，根本不知道退什么
用户第 3 轮：”那个多少钱来着” ← 指代消解：那个 = 订单里的商品
2.1 三种存储方案
方案一：LangGraph State + Checkpointer（推荐，新项目首选）
LangGraph 的 State 天然支持对话历史管理。核心机制：•Annotated[list, operator.add] → messages 字段使用 add reducer，每轮自动追加，不会覆盖•Checkpointer → 在轮次之间自动持久化 State（包括 messages）•thread_id → 标识同一个用户会话，同一个 thread_id 共享历史pythonfrom typing import Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver # 内存存储（开发用）
# from langgraph.checkpoint.postgres import PostgresSaver # 生产用
from langchain_core.messages import HumanMessage, AIMessage
# 1. State 里定义 messages 字段，用 add reducer 自动追加
class ChatState(TypedDict):
messages: Annotated[list, operator.add] # 每轮对话自动追加，不会覆盖
current_intent: str
# 2. 节点里直接从 state[“messages”] 读历史
def preprocess(state: ChatState) -> dict:
history = state[“messages”] # ← 这里就拿到了所有历史对话
current_msg = history[-1].content # 最新一条用户消息
# 把历史传给 LLM 做指代消解
rewrite_prompt = f”””
对话历史：
{format_history(history[:-1])}
用户最新输入：{current_msg}
请将最新输入改写为不依赖上下文也能理解的完整表达。
“””
rewritten = llm.invoke(rewrite_prompt)
return {“processed_input”: rewritten.content}
# 3. 编译时配置 Checkpointer → 自动持久化每轮状态
app = graph.compile(checkpointer=MemorySaver())
# 4. 每次调用带同一个 thread_id → 自动续接历史
config = {“configurable”: {“thread_id”: “user-session-123”}}
# 第 1 轮
app.invoke({“messages”: [HumanMessage(“查一下我的订单”)]}, config=config)
# 第 2 轮 → state[“messages”] 自动包含第 1 轮的内容
app.invoke({“messages”: [HumanMessage(“退了吧”)]}, config=config)
Checkpointer 的生产环境选择：
| Checkpointer | 适用环境 | 说明 |
|---|---|---|
MemorySaver | 开发/测试 | 内存存储，进程退出即丢失 |
SqliteSaver | 单机部署 | 文件级持久化，适合小规模 |
PostgresSaver | 生产环境 | 数据库持久化，支持多实例/高并发 |
方案二：外部存储（Redis / 数据库），手动管理
不用 LangGraph Checkpointer，自己管历史。适合已有会话管理系统的老项目：pythonimport redis
import json
r = redis.Redis()
def get_history(session_id: str) -> list:
“””从 Redis 读取对话历史”””
raw = r.lrange(f”chat:{session_id}”, 0, -1)
return [json.loads(msg) for msg in raw]
def save_message(session_id: str, role: str, content: str):
“””每轮对话结束后追加保存”””
r.rpush(f”chat:{session_id}”, json.dumps({“role”: role, “content”: content}))
r.expire(f”chat:{session_id}”, 3600 * 24) # 24 小时过期
# 在意图识别节点中使用
def classify_intent(state):
session_id = state[“session_id”]
history = get_history(session_id) # ← 手动从 Redis 读历史
current_input = state[“user_input”]
prompt = build_prompt(history, current_input) # 拼进 prompt
result = llm.invoke(prompt)
save_message(session_id, “user”, current_input)
save_message(session_id, “assistant”, result.content)
return {…}
方案三：前端传入（最简单，但有局限）
前端把对话历史完整传给后端，后端不存储：json{
“session_id”: “abc123”,
“messages”: [
{“role”: “user”, “content”: “查一下我的订单”},
{“role”: “assistant”, “content”: “好的，您的订单 ORD-001 已发货…”},
{“role”: “user”, “content”: “退了吧”}
]
}
简单，但历史长了之后请求体会很大，且前端可以篡改历史。只适合轻量场景。
2.2 三种方案对比
| LangGraph Checkpointer | 外部存储（Redis/DB） | 前端传入 | |
|---|---|---|---|
| 历史持久化 | 自动（Checkpointer 处理） | 手动（自己读写） | 不持久化 |
| 实现复杂度 | 低（框架内置） | 中（需要自己维护读写逻辑） | 最低 |
| 安全性 | 高（后端控制） | 高（后端控制） | 低（前端可篡改历史） |
| 多轮上限 | 需自己截断 | 需自己截断 | 受请求体大小限制 |
| 适合场景 | 新项目、LangGraph 体系 | 已有会话系统的老项目 | 轻量 Demo/短会话 |
| 生产可用 | 是 | 是 | 否（不建议） |
2.3 拿到历史后怎么传给 LLM
无论用哪种方案存储，最终传给 LLM 时的结构是一样的：pythondef build_messages_for_llm(summary: str, history: list, system_prompt: str) -> list:
“””构建最终传给 LLM 的 messages 列表”””
messages = []
# 1. System prompt（角色定义、规则等）
messages.append({“role”: “system”, “content”: system_prompt})
# 2. 如果有摘要（旧历史的压缩版），作为补充上下文
if summary:
messages.append({
“role”: “system”,
“content”: f”以下是之前对话的摘要，供你了解上下文：\n{summary}”
})
# 3. 对话历史（原文）
messages.extend(history)
return messages
# 最终传给 LLM 的结构：
# [
# {“role”: “system”, “content”: “你是一个客服助手…”},
# {“role”: “system”, “content”: “之前的摘要：用户咨询了蓝牙耳机…”}, ← 可选
# {“role”: “user”, “content”: “查一下我的订单”}, ← 历史
# {“role”: “assistant”, “content”: “好的，您的订单 ORD-001 已发货…”}, ← 历史
# {“role”: “user”, “content”: “退了吧”}, ← 当前输入
# ]
2.4 历史太长怎么办：策略总览
LLM 有上下文窗口限制，不可能把 200 轮对话全塞进去。常用策略：
| 策略 | 做法 | 适用场景 |
|---|---|---|
| 滑动窗口 | 只保留最近 N 轮（如最近 10 轮） | 大多数客服场景 |
| 摘要压缩 | 用 LLM 把旧历史压缩成一段摘要，只保留摘要 + 最近几轮 | 长对话场景 |
| 关键轮筛选 | 只保留包含关键实体/决策的历史轮次 | 复杂业务流程 |
实际项目不会只用一种策略，而是按对话轮数分级处理：对话进行中，历史不断增长
│
├── 阶段 1（< 10 轮）： 全量保留，不做任何处理
│
├── 阶段 2（10~30 轮）： 滑动窗口，只保留最近 N 轮
│
├── 阶段 3（> 30 轮）： 摘要压缩 + 滑动窗口
│ 旧历史压缩成摘要，近期保留原文
│
└── 阶段 4（超长会话）： 关键轮筛选 + 摘要 + 滑动窗口
只保留关键决策轮 + 摘要 + 近期原文
2.5 Token 预算计算
在做任何策略之前，先算清楚你有多少 token 可以分给历史。
一次 LLM 调用的 token 分配：模型上下文窗口（如 128K）
│
├── system prompt 约 500~2000 tokens（固定）
├── Few-shot 示例 约 500~1500 tokens（固定）
├── 当前用户输入 约 50~500 tokens（不可控）
├── ★ 对话历史 ？？？ tokens（这是你要管理的部分）
└── 预留给模型输出 约 500~2000 tokens（固定）
实际计算示例（以 GPT-4o / Claude Sonnet 为例）：上下文窗口：128,000 tokens
– system prompt： 1,000
– few-shot： 1,000
– 当前输入： 200
– 预留输出： 1,500
────────────────────────
可用于历史的预算： 124,300 tokens
但实际不建议用满！原因：
1. 长上下文注意力稀释，质量下降
2. token 成本线性增长
3. 响应延迟增加
实际建议预算：4,000 ~ 8,000 tokens 给历史（约 15~30 轮对话）
2.6 每次调用 LLM 前的完整处理流程
第一步：计算当前历史的 token 数
│
▼
第二步：是否超出预算？
│
├── 没超出 → 全量传入，不处理
│
└── 超出了 → 进入压缩流程
│
▼
第三步：分离历史
│
├── 近期历史（最近 6~10 轮）→ 保留原文
└── 远期历史（更早的部分） → 进入压缩
│
▼
第四步：压缩远期历史
│
├── 已有摘要？→ 基于旧摘要 + 新的远期历史，更新摘要
└── 没有摘要？→ 首次生成摘要
│
▼
第五步：拼接最终历史
[摘要] + [近期原文]
│
▼
第六步：再次检查 token 数
│
├── 仍然超出 → 缩小近期窗口 / 精简摘要
└── 合格 → 传入 LLM
2.7 Token 计数方法
pythonimport tiktoken
# 选择与你使用的模型匹配的编码器
encoder = tiktoken.encoding_for_model(“gpt-4o”)
# Claude 模型没有官方 tiktoken，可以用近似估算
# 中文：1 个汉字 ≈ 1.5~2 tokens
# 英文：1 个单词 ≈ 1~1.5 tokens
def count_tokens(messages: list[dict]) -> int:
“””计算消息列表的 token 数”””
total = 0
for msg in messages:
total += len(encoder.encode(msg[“content”]))
total += 4 # 每条消息的格式开销（role 标签等）
return total
def count_tokens_approx(text: str) -> int:
“””无 tiktoken 时的近似估算（中文场景）”””
return int(len(text) * 1.5)
2.8 策略一：滑动窗口
最简单的方案，只保留最近 N 轮：pythondef trim_history(messages: list, max_turns: int = 10) -> list:
“””滑动窗口：只保留最近 N 轮”””
if len(messages) <= max_turns * 2: # 每轮 = user + assistant
return messages
return messages[-(max_turns * 2):]
2.9 策略二：摘要压缩
核心思路
把老的对话历史交给 LLM，让它压缩成一段简短的摘要。后续调用时，用”摘要 + 最近几轮原文”代替完整历史。原始历史（40 轮，8000 tokens）：
第 1 轮：用户问了产品价格
第 2 轮：客服报价 299 元
第 3 轮：用户问了颜色选择
…
第 38 轮：用户确认收货地址
第 39 轮：客服确认订单
第 40 轮：用户说”那个快递能加急吗” ← 当前轮
压缩后（摘要 + 近 6 轮，约 1500 tokens）：
[摘要]：用户咨询了产品（蓝牙耳机，299元），选择了黑色款，
已下单（订单号 ORD-456），收货地址为北京市海淀区xx路。
第 35~40 轮：原文保留
什么时候触发压缩
策略 A：按 token 阈值触发（推荐）pythonMAX_HISTORY_TOKENS = 6000 # 历史 token 预算
RECENT_TURNS_TO_KEEP = 6 # 近期保留的轮数
SUMMARY_TRIGGER = 4000 # 超过这个数就触发压缩
def should_compress(messages: list, existing_summary: str) -> bool:
history_tokens = count_tokens(messages)
summary_tokens = count_tokens_approx(existing_summary) if existing_summary else 0
return (history_tokens + summary_tokens) > SUMMARY_TRIGGER
策略 B：按轮数触发pythonMAX_RAW_TURNS = 10 # 原文最多保留 10 轮，超了就压缩
def should_compress(messages: list) -> bool:
turn_count = len([m for m in messages if m[“role”] == “user”])
return turn_count > MAX_RAW_TURNS
摘要生成的 Prompt
首次生成摘要：你是一个对话摘要助手。请将以下客服对话历史压缩为一段简洁的摘要。
要求：
1. 保留所有关键事实：用户身份、涉及的产品/订单/金额、已做出的决策、已确认的信息
2. 保留所有未解决的问题和待办事项
3. 丢弃寒暄、重复确认、无实质内容的轮次
4. 使用第三人称描述（”用户…”、”客服…”）
5. 控制在 200 字以内
对话历史：
{old_messages}
摘要：
追加更新摘要（历史继续增长时，基于旧摘要 + 新轮次更新）：你是一个对话摘要助手。请基于已有摘要和新增的对话内容，更新摘要。
要求：
1. 将新增对话中的关键信息合并到摘要中
2. 如果新信息与旧摘要有冲突（如用户改了地址），以新信息为准
3. 删除已经过时或不再相关的旧信息
4. 控制在 300 字以内
已有摘要：
{existing_summary}
新增对话：
{new_messages}
更新后的摘要：
摘要存在哪里
| 存储方式 | 做法 | 适用场景 |
|---|---|---|
| LangGraph State | 在 State 中加一个 summary: str 字段，Checkpointer 自动持久化 | LangGraph 项目 |
| Redis | chat_summary:{session_id} 单独存储 | 已有 Redis 的项目 |
| 数据库 | sessions 表加一个 summary 字段 | 需要长期保留 |
LangGraph 中的做法：pythonclass ChatState(TypedDict):
messages: Annotated[list, operator.add] # 对话历史
summary: str # 压缩摘要
def maybe_compress(state: ChatState) -> dict:
messages = state[“messages”]
ifnot should_compress(messages, state[“summary”]):
return {} # 不需要压缩
# 分离：近期保留原文，远期压缩
recent = messages[-(RECENT_TURNS_TO_KEEP * 2):] # 最近 N 轮
old = messages[:-(RECENT_TURNS_TO_KEEP * 2)] # 更早的部分
# 生成/更新摘要
if state[“summary”]:
new_summary = llm.invoke(update_summary_prompt(state[“summary”], old))
else:
new_summary = llm.invoke(create_summary_prompt(old))
# 用摘要 + 近期原文替换完整历史
return {
“summary”: new_summary.content,
“messages”: recent, # 只保留近期，旧的被摘要替代
}
摘要怎么传给后续的 LLM 调用
在构建 prompt 时，摘要作为 system message 插入，近期对话原文紧随其后：pythondef build_messages_for_llm(state: ChatState) -> list:
messages = []
# 1. System prompt
messages.append({
“role”: “system”,
“content”: “你是一个客服助手…”
})
# 2. 如果有摘要，作为上下文补充
if state[“summary”]:
messages.append({
“role”: “system”,
“content”: f”以下是之前对话的摘要，供你了解上下文：\n{state[‘summary’]}”
})
# 3. 近期对话原文
messages.extend(state[“messages”])
return messages
摘要的质量怎么保证
| 风险 | 后果 | 应对措施 |
|---|---|---|
| 摘要丢失关键信息 | 后续对话出现”失忆”，用户需要重复说 | 在 prompt 中明确列出必须保留的信息类型 |
| 摘要出现幻觉 | 摘要里出现用户从没说过的信息 | 要求摘要只能包含对话中明确出现的事实 |
| 摘要越来越长 | 多次追加更新后摘要膨胀 | 设置字数上限（如 300 字），超出时要求 LLM 精简 |
| 摘要更新时丢失旧信息 | 旧摘要中的信息在更新时被遗忘 | 使用”追加更新”模式而非”重新生成” |
◆
2.10 策略三：关键轮筛选
核心思路
不是所有对话都同等重要。"好的""嗯""收到" 这些轮次没有信息量。关键轮筛选就是给每一轮打分，只保留高价值的轮次。
什么样的轮次是”关键轮”
| 关键程度 | 特征 | 示例 |
|---|---|---|
| 必须保留 | 包含业务实体（订单号、金额、产品名） | “我的订单号是 ORD-456，花了 299 元” |
| 必须保留 | 用户做出了明确决策 | “我要退款””就选黑色的” |
| 必须保留 | 系统做出了承诺 | “已为您创建退款工单，3 个工作日内到账” |
| 必须保留 | 包含问题定义或需求描述 | “手机充电到 80% 就不动了” |
| 可以保留 | 包含情绪表达（用于情绪分析） | “等了一周了真的很失望” |
| 可以丢弃 | 纯确认/应答 | “好的””嗯””收到了””谢谢” |
| 可以丢弃 | 重复内容 | 用户反复描述同一个问题 |
| 可以丢弃 | 寒暄 | “你好””在吗””打扰了” |
方案 A：规则打分（快、免费，适合大多数场景）
pythondef score_turn(user_msg: str, assistant_msg: str) -> float:
“””
给一轮对话打分，0.0~1.0。
分数越高越关键，越应该保留。
“””
score = 0.0
combined = user_msg + assistant_msg
# —- 加分项 —-
# 包含业务实体 → 高价值
entity_patterns = [
r”订单[号#]?\s*[:：]?\s*\w+”, # 订单号
r”\d+(\.\d+)?\s*元”, # 金额
r”1[3-9]\d{9}”, # 手机号
r”\d{4}[-/]\d{1,2}[-/]\d{1,2}”, # 日期
]
for pattern in entity_patterns:
if re.search(pattern, combined):
score += 0.3
# 包含决策动词 → 高价值
decision_words = [“要退”, “确认”, “同意”, “取消”, “下单”, “选择”, “不要”, “换成”]
if any(w in combined for w in decision_words):
score += 0.3
# 包含系统承诺 → 高价值
commitment_words = [“已为您”, “工单”, “将在”, “预计”, “已创建”, “已提交”]
if any(w in assistant_msg for w in commitment_words):
score += 0.3
# 消息够长，信息密度可能高 → 适度加分
if len(user_msg) > 30:
score += 0.1
# —- 减分项 —-
# 纯短回复 → 低价值
if len(user_msg) <= 4: # “好的””嗯嗯””收到”
score -= 0.3
# 寒暄 → 低价值
greetings = [“你好”, “在吗”, “谢谢”, “感谢”, “辛苦”, “拜拜”, “再见”]
if any(g in user_msg for g in greetings) and len(user_msg) < 10:
score -= 0.3
return max(0.0, min(1.0, score))
def filter_key_turns(
messages: list[dict],
keep_ratio: float = 0.4, # 保留比例
min_score: float = 0.2, # 最低分数线
always_keep_last_n: int = 6, # 最近 N 轮无条件保留
) -> list[dict]:
“””
筛选关键轮次。
保留：高分轮次 + 最近 N 轮（无条件保留）。
“””
# 按 user/assistant 配对
turns = []
i = 0
while i < len(messages) – 1:
if messages[i][“role”] == “user” and messages[i + 1][“role”] == “assistant”:
turns.append({
“user”: messages[i],
“assistant”: messages[i + 1],
“score”: score_turn(messages[i][“content”], messages[i + 1][“content”]),
“index”: len(turns),
})
i += 2
else:
i += 1
ifnot turns:
return messages
total = len(turns)
# 最近 N 轮无条件保留
recent_start = max(0, total – always_keep_last_n)
recent_turns = turns[recent_start:]
older_turns = turns[:recent_start]
# 从旧轮次中筛选高分的
keep_count = max(1, int(len(older_turns) * keep_ratio))
scored_old = [t for t in older_turns if t[“score”] >= min_score]
scored_old.sort(key=lambda t: t[“score”], reverse=True)
kept_old = scored_old[:keep_count]
kept_old.sort(key=lambda t: t[“index”]) # 恢复时间顺序
# 拼接：保留的旧轮次 + 全部近期轮次
result = []
for t in kept_old + recent_turns:
result.append(t[“user”])
result.append(t[“assistant”])
return result
方案 B：LLM 筛选（更准，但多一次调用）
让 LLM 自己判断哪些轮次重要：你是一个对话分析助手。以下是一段客服对话历史，共 {n} 轮。
请筛选出最关键的轮次（保留原文，不要改写）。
筛选标准：
1. 必须保留：包含订单号、金额、产品名等业务实体的轮次
2. 必须保留：用户做出决策（退款、下单、确认）的轮次
3. 必须保留：客服做出承诺（创建工单、预计时间）的轮次
4. 必须保留：首次描述问题的轮次
5. 可以丢弃：纯寒暄、简单确认（好的/嗯/收到）、重复描述
输出格式：返回保留轮次的编号列表，如 [1, 3, 5, 8, 9, 10]
对话历史：
{numbered_messages}
保留的轮次编号：
关键轮筛选 vs 摘要压缩的区别
| 摘要压缩 | 关键轮筛选 | |
|---|---|---|
| 输出形式 | 一段自然语言摘要 | 原始对话轮次（保留原文） |
| 信息损失 | 有（LLM 可能遗漏细节） | 低（保留的轮次是原文） |
| 适合场景 | 历史极长（>30 轮）、只需要大致上下文 | 历史中等长（10~30 轮）、需要精确引用 |
| 成本 | 一次 LLM 调用 | 规则方案免费 / LLM 方案一次调用 |
| 风险 | 摘要幻觉 | 可能误删重要轮次 |
2.11 实际项目推荐：三段式混合使用
对话历史
│
├── 最近 6 轮 → 原文保留（不动）
│
├── 7~30 轮 → 关键轮筛选（保留高分轮次原文）
│
└── 30 轮以前 → 摘要压缩（压缩成一段文字）
最终传给 LLM 的历史 = [摘要] + [筛选的关键旧轮] + [最近 6 轮原文]
具体实现：pythonRECENT_WINDOW = 6 # 最近 N 轮保留原文
KEY_TURN_WINDOW = 30 # 关键轮筛选的范围
MAX_HISTORY_TOKENS = 6000 # 历史总 token 预算
def prepare_history(
all_messages: list[dict],
existing_summary: str,
) -> tuple[str, list[dict]]:
“””
处理对话历史，返回 (摘要, 保留的消息列表)。
三段式：摘要 + 关键旧轮 + 近期原文。
“””
total_turns = len(all_messages) // 2
# — 分段 —
recent_msgs = all_messages[-(RECENT_WINDOW * 2):] # 最近 6 轮
if total_turns <= RECENT_WINDOW:
# 对话很短，不需要任何处理
return “”, all_messages
if total_turns <= KEY_TURN_WINDOW:
# 中等长度：关键轮筛选 + 近期
older_msgs = all_messages[:-(RECENT_WINDOW * 2)]
key_msgs = filter_key_turns(older_msgs, keep_ratio=0.4)
return existing_summary, key_msgs + recent_msgs
# 超长对话：摘要 + 关键轮 + 近期
mid_msgs = all_messages[-(KEY_TURN_WINDOW * 2):-(RECENT_WINDOW * 2)] # 中间段
old_msgs = all_messages[:-(KEY_TURN_WINDOW * 2)] # 最老段
# 最老段压缩成摘要
new_summary = update_or_create_summary(existing_summary, old_msgs)
# 中间段做关键轮筛选
key_msgs = filter_key_turns(mid_msgs, keep_ratio=0.3)
# 检查 token 是否超预算
total_tokens = (
count_tokens_approx(new_summary)
+ count_tokens(key_msgs)
+ count_tokens(recent_msgs)
)
if total_tokens > MAX_HISTORY_TOKENS:
# 超预算：进一步压缩，减少关键轮保留比例
key_msgs = filter_key_turns(mid_msgs, keep_ratio=0.2, min_score=0.4)
return new_summary, key_msgs + recent_msgs
2.12 本章小结
模型没有记忆，对话历史是你存的、你拼的、你传的。 LangGraph 的 State + Checkpointer 是目前最省心的方案——你只管往 messages 里追加，它替你持久化和续接。历史太长时，用滑动窗口、摘要压缩、关键轮筛选三种手段控制传入量，生产环境推荐三段式混合使用。◆
三、生产环境完整配置建议
3.1 参数配置（做成可调的，不要硬编码）
yaml# config.yaml – 对话历史管理配置
history:
# token 预算
max_history_tokens: 6000
# 滑动窗口
recent_window: 6 # 最近 N 轮无条件保留
# 摘要压缩
summary_trigger_turns: 20 # 超过 N 轮触发压缩
summary_max_chars: 500 # 摘要最大字数
summary_model: “gpt-4o-mini” # 摘要用便宜的小模型即可
# 关键轮筛选
key_turn_keep_ratio: 0.4 # 保留比例
key_turn_min_score: 0.2 # 最低分数线
key_turn_method: “rule” # rule（规则打分） 或 llm（LLM 筛选）
3.2 不同业务场景的推荐配置
| 场景 | 近期窗口 | 压缩触发 | 保留比例 | 说明 |
|---|---|---|---|---|
| 售前客服 | 6 轮 | 15 轮 | 0.3 | 对话短，重点在产品信息 |
| 售后/工单 | 10 轮 | 20 轮 | 0.5 | 决策多，需要更多上下文 |
| 技术支持 | 8 轮 | 25 轮 | 0.5 | 问题描述长，排查步骤重要 |
| 闲聊/陪伴 | 4 轮 | 10 轮 | 0.2 | 信息密度低，积极压缩 |
| 金融/法律 | 15 轮 | 30 轮 | 0.7 | 高风险，尽量多保留 |
3.3 监控指标
上线后需要关注的指标：
| 指标 | 怎么测 | 警戒线 |
|---|---|---|
| 历史截断导致的”失忆”率 | 用户重复描述已说过的信息的比例 | > 5% 需要调大窗口 |
| 摘要质量 | 抽样人工评审摘要是否丢失关键信息 | 丢失率 > 10% 需要优化 prompt |
| 平均历史 token 数 | 每次调用传入的历史 token 统计 | 持续超预算需要收紧策略 |
| 压缩触发频率 | 多少比例的请求触发了压缩 | 过高说明窗口设小了 |
声明：来自HG全栈小课堂，仅代表创作者观点。链接：https://eyangzhen.com/8404.html
