如果你玩过tavo，这一节就会很好理解 (tavo就是sillytarvern的安卓版本，只不过是闭源。)

tavo是一个角色扮演游戏，大模型通过读取角色卡和世界书的内容，扮演角色，或者扮演RPG（一个卡里面有多个角色）。

tavo里面的提示词基本是如下结构拼接的：
用户当前输入 +角色卡+历史上下文（用户输入+大模型返回。一共12轮，用户输入一次+大模型返回一次为一轮）+长记忆（目前看起来是全量输入）。长记忆在超出大模型TOKEN数之后会用RAG进行关键记忆提取。如果扫描到有世界书的提示词，则会注入在长记忆之前。

换句话说，所谓长记忆，或者上下文，都是一系列的内容拼接起来的。拼接内容是按照一定的预设格式进行输入，同时，拼接大小是满足模型输入最大信息来确定的。总体要求是不能超过。输入最大信息。

这里插个题外话，我自己原来想的是，上下文越长的，玩起来越有沉浸感。结果我发现每次大模型的输入只有几万token，基本上市面的普通模型输入长度都可以满足。沉浸式体验我发现性价比最高的，只有一个：gemini 3 flash preview。 输出长度合适并且很能满足用户体验。deepseek等国内模型，太死板，非得按照原版剧情玩。玩tavo的神级模型有且只有一个，那就是claude sonnet 4.6。虽然很贵，但是高潮部分用这个准没错。gemini 3 flash preview有一个问题就是用时间长了文风就锁死了。可以多模型混合使用可以解决这个问题。另外国内的qwen卡输入太严重，一旦有一些不合适的字眼进入了长记忆，基本就用不了了。
另外，长记忆提示词可以做一些修改，因为原版长记忆提示词只记录用户的行为，关键NPC出现了明显变化则记不住，经常导致很久之前把A NPC杀了或者怎样，后面玩几轮这个A NPC又复活了。因此需要手动调试长记忆提示词。

TAVO现在从某种程度成了我的赛博老家，哈哈。

下面继续讲上下文管理机制。langchain的上下文输入和tavo路子基本一致。也是一堆提示词拼接而成。

System → History → Knowledge → User Query 这个是大模型RAG 提示词的拼接顺序

# 1. 核心依赖（记忆专用）
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 2. 初始化模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) #### 如何使用api key调用。请看第一节

# ==============================================
# 【第一步】定义带「记忆占位符」的提示词（对标TaVo的记忆位置）
# ==============================================
prompt = ChatPromptTemplate.from_messages([
    # 1. 系统提示词
    ("system", "你是专业AI助手，记住用户的所有对话信息"),
    # 2. 记忆占位符
    MessagesPlaceholder(variable_name="history"),
    # 3. 用户当前问题
    ("user", "{question}")
])
# ==============================================
# 【第二步】基础LCEL链（不带记忆）
# ==============================================
base_chain = prompt | llm | StrOutputParser()
# ==============================================
# 【第三步】给链包装「记忆」→ 核心！
# ==============================================
# 内存存储记忆（生产环境用Redis/数据库/或RAG的向量数据库）
store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# 最终带记忆的链
chain_with_memory = RunnableWithMessageHistory(
    base_chain,                # 你的基础链
    get_session_history,       # 记忆获取函数
    input_messages_key="question",  # 输入数据的key
    history_messages_key="history" # 提示词里的记忆占位key
)
# ==============================================
# 【测试：多轮对话】
# ==============================================
if __name__ == "__main__":
    # 会话ID：区分不同用户（TaVo 不同聊天窗口）
    config = {"configurable": {"session_id": "user_123"}}
    # 第一轮
    print("第一轮：", chain_with_memory.invoke({"question": "我叫小明"}, config=config))
    # 第二轮（AI会记住你叫小明）
    print("第二轮：", chain_with_memory.invoke({"question": "我叫什么？"}, config=config))

有几个问题需要说明：

prompt = ChatPromptTemplate.from_messages([
    # 1. 系统提示词
    ("system", "你是专业AI助手，记住用户的所有对话信息"),
    # 2. 记忆占位符
    MessagesPlaceholder(variable_name="history"),
    # 3. 用户当前问题
    ("user", "{question}")
])
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

我们都知道，session_id的作用是区分会话的。下面的InMemoryChatMessageHistory取会话是使用session_id，但MessagesPlaceholder存会话，却不用session_id，这是为什么？
MessagesPlaceholder(variable_name=“history”), 这个就是个空白的格式模板，没有任何用户数据，没有 session_id。
调用链的时候，传入了session_id:

# 调用用户1
config1 = {"configurable": {"session_id": "用户1"}}
chain_with_memory.invoke({"question": "我叫小明"}, config=config1)
# 调用用户2
config2 = {"configurable": {"session_id": "用户2"}}
chain_with_memory.invoke({"question": "我叫小红"}, config=config2)

即：调用的时候进行传入，而不是在提示词传入。
因此，可以从InMemoryChatMessageHistory按照session_id 进行内容获取。

chain_with_memory 这个相当于做了一个带记忆的链，也是可以用chain1 | model | perser进行拼接。

但有个很致命的问题：如果对话长度太大，超出了大模型的最大输入上下文，怎么办？

InMemoryChatMessageHistory 是无限制存储的，生产环境不能用。

因此，一般需要限制条数或者限制token数来卡提示词大小。

import os
from langchain_core.runnables import RunnablePassthrough
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import trim_messages
from langchain_core.messages.utils import count_tokens_approximately
from langchain_core.chat_history import InMemoryChatMessageHistory

# ================== 配置（请替换为你的有效 API Key） ==================
API_KEY = "sk-XXXXXXXXX"
BASE_URL = "https://api.deepseek.com"   # 或 OpenAI 的地址

model = ChatOpenAI(
    model="deepseek-chat",
    openai_api_key=API_KEY,
    openai_api_base=BASE_URL,
    temperature=0.7
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是专业助手，请简洁回答。"),
    MessagesPlaceholder(variable_name="history"),
    ("user", "{question}")
])   # 拼prompt，即系统提示词+历史记忆+用户提示词。

base_chain = prompt | model | StrOutputParser()

# 会话存储
store = {}
def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]    # 按照session_id去找记忆。一般session_id是用户ID+对话ID来做用户间隔离和对话间隔离。

# ========== 1. 按消息条数截断（保留最近 4 条消息 = 2 轮对话） ==========
trimmer_by_count = trim_messages(
    max_tokens=4,
    strategy="last",
    token_counter=len,          # 每条消息计数为 1
    allow_partial=False,
    include_system=False,
)   

chain_with_count_trim = RunnablePassthrough.assign(
    history=lambda inputs: trimmer_by_count.invoke(inputs["history"])
) | base_chain   #根据trimmer_by_count进行历史记录裁剪

chain_count = RunnableWithMessageHistory(
    chain_with_count_trim,
    get_session_history,
    input_messages_key="question",
    history_messages_key="history"
)  #创建记忆链


# 下同
# ========== 2. 按 Token 数量截断（保留总 token ≤ 50，使用估算函数） ==========
trimmer_by_token = trim_messages(
    max_tokens=50,
    strategy="last",
    token_counter=count_tokens_approximately,
    allow_partial=False,
)

chain_with_token_trim = RunnablePassthrough.assign(
    history=lambda inputs: trimmer_by_token.invoke(inputs["history"])
) | base_chain

chain_token = RunnableWithMessageHistory(
    chain_with_token_trim,
    get_session_history,
    input_messages_key="question",
    history_messages_key="history"
)

# ================== 测试对话 ==================
if __name__ == "__main__":
    config = {"configurable": {"session_id": "test_user"}}

    print("===== 测试按消息条数截断（保留最近4条） =====")
    # 连续对话 4 轮（8 条消息），但只保留最后 4 条
    chain_count.invoke({"question": "我叫张三"}, config=config)
    chain_count.invoke({"question": "我20岁"}, config=config)
    chain_count.invoke({"question": "我喜欢篮球"}, config=config)
    chain_count.invoke({"question": "我家养了只猫"}, config=config)

    # 问名字，由于“我叫张三”已经被截断（因为超过4条），模型应该不知道
    response = chain_count.invoke({"question": "我叫什么名字？"}, config=config)
    print(f"AI: {response}")

    # 重新开始一个新的 session 测试按 token 截断
    print("\n===== 测试按 Token 截断（≤50 token） =====")
    config2 = {"configurable": {"session_id": "test_user2"}}
    # 先发送一条较长的问题，让 token 快速累积
    chain_token.invoke({"question": "请详细介绍一下什么是大语言模型，包括它的原理、训练方式、应用场景。"}, config=config2)
    chain_token.invoke({"question": "我刚才问的是什么？"}, config=config2)
    response2 = chain_token.invoke({"question": "再重复一遍我的第一个问题"}, config=config2)
    print(f"AI: {response2}")

其中：

trimmer_by_count = trim_messages(
    max_tokens=4,
    strategy="last",
    token_counter=len,          # 每条消息计数为 1
    allow_partial=False,
    include_system=False,
)   

这个设计，我个人不喜欢，因为这里理解有歧义，并且需要程序员严格自我定义好裁剪单位：token数或者是消息数。个人认为不如用个字典或者什么数据类型包一下比较好。

max_tokens 这个定义很容易歧义，就是最大token数。但后面的赋值代表两种含义：token数或者是条目数，赋值都是整数，如何区分：

我们看一下官方文档：

token_counter: Function or llm for counting tokens in a BaseMessage or a list of
BaseMessage. If a BaseLanguageModel is passed in then
BaseLanguageModel.get_num_tokens_from_messages() will be used.
Set to len to count the number of messages in the chat history.

        .. note::
            Use `count_tokens_approximately` to get fast, approximate token counts.
            This is recommended for using `trim_messages` on the hot path, where
            exact token counting is not necessary.

换句话说，只要token_counter的后面为len就是信息条数，非len就是计算token数，如果想要快速计算token就用count_tokens_approximately。也可以用其他大模型自带的tokenizer来算。

实际生产来说，肯定不会用langchain自己的记录，而更倾向于用FAISS或者Milvus 这种向量数据库。因为langchain绝大多数一定是跟RAG相结合。

本节分享完毕。