【Bug已解决】Public API for safe message retention cutoff 解决方案一、现象长什么样在langchain里做消息保留裁剪把超长对话截断到模型窗口内时用户没有一个稳定、公开、语义清晰的 API 可用只能自己手写循环去 pop 消息于是各种翻车有人直接messages messages[-N:]结果把最前面的SystemMessage切掉了模型瞬间失忆人设有人用trim_messages但参数配错strategylast把最早的用户问题丢了allow_partialFalse把一条长消息整条砍掉留下半截语义多人各自实现裁剪逻辑行为不一致A 的服务保留 systemB 的服务不保留导致同一个 chain 在不同项目里表现不同当消息里混有ToolMessage、AIMessage带 tool_calls、HumanMessage时随意截断会破坏tool_call 必须配对 tool_result的约束触发 provider 400没有统一的安全裁剪契约代码评审也没法判断这个 pop 对不对只能靠人肉。一句话缺少一个公开、安全、默认就正确的消息保留裁剪 API导致大家各写各的、反复踩配对与系统消息丢失的坑。二、背景对话型应用必然面对上下文窗口有限这个现实。随着多轮对话变长必须把旧消息裁掉。但裁剪不是简单的保留最后 N 条它要满足几条硬约束SystemMessage 通常必须保留人设、指令在最前Tool 调用必须成对如果保留了一个带tool_calls的AIMessage它对应的ToolMessage也必须保留否则 provider 报错不能把一条消息劈成两半除非显式允许allow_partial裁剪后总 token 须 ≤ 预算。langchain-core提供了trim_messages但它参数多、语义细新手容易用错而且社区一直希望有一个更上层的、开箱即安全的封装把上面四条约束默认做对而不是每次都让用户去调参。三、根因根因是**安全的默认没有被封装成公开 API**用户被迫直接面对底层、易错的trim_messages# 用户手写极易出错 def keep_recent(messages, n10): return messages[-n:] # 直接把 SystemMessage 干掉且不考虑 tool 配对# trim_messages 默认不对 system 做保护参数配错就翻车 trimmed trim_messages( messages, max_tokens2000, strategylast, # 忘了 token_counter、忘了 include_system行为不可预期 )真正安全的裁剪需要考虑消息类型拓扑。langchain当时没有一个把常见正确姿势固化下来的公开顶层函数于是每个人重新发明轮子重复踩坑。四、最小可运行复现from langchain_core.messages import ( SystemMessage, HumanMessage, AIMessage, ToolMessage, ) def unsafe_keep_recent(messages, n3): # 典型错误实现只保留最后 3 条 return messages[-n:] history [ SystemMessage(content你是客服助手), HumanMessage(content我的订单在哪), AIMessage(content, tool_calls[{id: c1, name: query_order}]), ToolMessage(content订单已发货, tool_call_idc1), HumanMessage(content那什么时候到), ] kept unsafe_keep_recent(history, n3) print([type(m).__name__ for m in kept]) # - [ToolMessage, HumanMessage, HumanMessage] # 问题SystemMessage 没了且 AIMessage(tool_calls) 被删但 ToolMessage 留下 # 配对断裂更早的用户问题也丢了。运行后可以看到系统消息丢失、tool 配对断裂——这正是线上 400 的来源。五、解决方案第一层最小直接修复最小修复是用trim_messages给出一个安全默认封装固化保留 system 不劈半 双向策略from langchain_core.messages import trim_messages, SystemMessage from langchain_core.messages.utils import count_tokens_approximately def safe_retain(messages, *, max_tokens: int 2000): return trim_messages( messages, max_tokensmax_tokens, strategylast, token_countercount_tokens_approximately, include_systemTrue, # 永远保留 system allow_partialFalse, # 不劈半条消息 start_onhuman, # 从人类消息开始裁避免砍断 tool 链 )这样即使用户只调用safe_retain(history)也能得到一个系统消息在、tool 链完整、不超 token的结果。六、解决方案第二层结构化改进把安全保留规则做成可配置的策略对象作为唯一事实来源并内置配对校验from dataclasses import dataclass, field from typing import Callable, List, Optional dataclass(frozenTrue) class LangChainMessageRetentionPolicy: 消息安全保留策略在预算内裁剪并满足硬约束。 约束 1. SystemMessage 始终保留除非 explicitly drop 2. 带 tool_calls 的 AIMessage 必须与其 ToolMessage 成对保留 3. allow_partialFalse 时不劈半条消息 max_tokens: int 2000 keep_system: bool True allow_partial: bool False token_counter: Callable field(defaultcount_tokens_approximately) def retain(self, messages: List) - List: trimmed trim_messages( messages, max_tokensself.max_tokens, strategylast, token_counterself.token_counter, include_systemself.keep_system, allow_partialself.allow_partial, start_onhuman, ) self._assert_tool_pairs(trimmed) return trimmed staticmethod def _assert_tool_pairs(messages: List) - None: ai_ids { tc[id] for m in messages if hasattr(m, tool_calls) and m.tool_calls for tc in m.tool_calls } tool_ids { m.tool_call_id for m in messages if hasattr(m, tool_call_id) } # 若保留了 AI 的 tool_call必须有对应 ToolMessage orphan ai_ids - tool_ids if orphan: raise ValueError(f裁剪后 tool_call 配对断裂缺失: {orphan}) def demo() - None: from langchain_core.messages import ( SystemMessage, HumanMessage, AIMessage, ToolMessage, ) history [ SystemMessage(content你是客服), HumanMessage(content订单在哪), AIMessage(content, tool_calls[{id: c1, name: q}]), ToolMessage(content已发货, tool_call_idc1), HumanMessage(content何时到), ] policy LangChainMessageRetentionPolicy(max_tokens2000) print([type(m).__name__ for m in policy.retain(history)]) if __name__ __main__: demo()七、解决方案第三层断言 / CI 守护用测试锁死安全约束from langchain_core.messages import ( SystemMessage, HumanMessage, AIMessage, ToolMessage, ) import pytest from your_module import LangChainMessageRetentionPolicy def test_system_message_always_kept(): policy LangChainMessageRetentionPolicy(max_tokens50, keep_systemTrue) msgs [SystemMessage(contentsys), HumanMessage(contentx * 200)] out policy.retain(msgs) assert any(isinstance(m, SystemMessage) for m in out) def test_tool_pair_not_broken(): policy LangChainMessageRetentionPolicy(max_tokens2000) msgs [ SystemMessage(contents), HumanMessage(contentq), AIMessage(content, tool_calls[{id: c1, name: q}]), ToolMessage(contentr, tool_call_idc1), ] # 不会抛 orphan 错误 out policy.retain(msgs) assert any(m.tool_call_id c1 for m in out) def test_orphan_tool_call_raises(): policy LangChainMessageRetentionPolicy(max_tokens2000) # 构造一个会断裂的场景删掉 ToolMessage 后保留 AI msgs [ AIMessage(content, tool_calls[{id: c1, name: q}]), ] with pytest.raises(ValueError): policy.retain(msgs)把safe_retain/LangChainMessageRetentionPolicy作为公开 API 暴露文档里明确推荐它替代手写messages[-N:]。八、排查清单你是否直接messages[-N:]截断若是system 消息可能已丢失。裁剪后是否仍满足 tool_call 与 tool_result 配对可用上面的断言校验。是否保留 SystemMessageinclude_systemTrue了吗是否出现半截消息allow_partial是否符合预期总 token 是否真的 ≤ 预算用token_counter复核。是否所有项目都统一用同一个安全裁剪 API而非各写各的九、小结消息保留裁剪是个看似简单、实则布满约束的活系统消息不能丢、tool 调用必须配对、消息不能劈半、token 不能超预算。没有公开的安全 API 时大家各写各的messages[-N:]反复踩坑。trim_messages给出了底层能力但参数易错正确做法是把它固化成一个默认就安全的LangChainMessageRetentionPolicy保留 system、配对校验、不劈半并写进文档作为推荐路径再用 pytest 把几条硬约束锁死让裁剪这件事开箱即正确。