Langchain：消息历史长度限制问题排查与技术咨询

想要限制消息历史仅保留最后N条，避免大语言模型（LLM）过载。计划结合使用RunnableWithMessageHistory与过滤函数，但遇到两个问题：

1. 限制函数无法正常工作
2. 实际用户消息无法传入模型

1. 为何RunnablePassthrough未对消息历史生效？
2. 结合input_messages_key使用HumanMessage("{user_message_key}")的方式是否正确？了解有HumanMessagePromptTemplate，但不确定为何必须使用它。
3. 是否可以在get_session_history中应用_filter_messages？其优缺点是什么？

from typing import List, Union
from langchain_openai import ChatOpenAI
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables import RunnablePassthrough


store = {}
model_name = "gpt-3.5-turbo"
system_message = "You're a helpful, friendly AI"
model = ChatOpenAI(model=model_name)

# Function to get chat message history
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    history: BaseChatMessageHistory = store[session_id]
    # history.messages = self._filter_messages(history.messages)
    return history

def filter_messages(messages: List[Union[HumanMessage, AIMessage]]) -> List[Union[HumanMessage, AIMessage]]:
    return messages[-5:]

def process_text(text: str, session_history: BaseChatMessageHistory, session_id: str) -> AIMessage:
    # Create the prompt template
    prompt = ChatPromptTemplate.from_messages(
        [
            SystemMessage(system_message),
            MessagesPlaceholder(variable_name="chat_hist"),
            HumanMessage("{user_message_key}")
        ]
    )

    runnable = RunnablePassthrough.assign(chat_hist=lambda x: filter_messages(x["chat_hist"])) | prompt | model

    with_message_history = RunnableWithMessageHistory(
        runnable=runnable,
        get_session_history=get_session_history,
        history_messages_key="chat_hist",
        input_messages_key="usr_msg",
    )

    resp: AIMessage = with_message_history.invoke(
        {"usr_msg": [HumanMessage(content=text)]},
        config={"configurable": {"session_id": session_id}},
    )
    print(resp.content)
    return resp


session_id = "test_session"
previous_messages = [
    HumanMessage(content="I love the color Pink. It's my favorite color."),
    AIMessage(content="That's great to know!"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Hello, my name is John."),
    AIMessage(content="Hi John, great to meet you."),
    HumanMessage(content="I'm sick of this rain."),
]
session_history = get_session_history(session_id)
for message in previous_messages:
    session_history.add_message(message)

user_input = "Repeat exactly: GOODBYE"
response:AIMessage = process_text(user_input, session_history, session_id)
# answer: I'm sorry to hear that you're feeling down because of the rain. Is there anything I can do to help cheer you up?

1. RunnablePassthrough未生效的原因 #

代码中RunnablePassthrough引用的x["chat_hist"]是RunnableWithMessageHistory注入的原始历史消息，但存在两个核心问题：

• 过滤逻辑仅在当前调用的runnable链中临时生效，并未修改存储的历史记录；
• prompt中HumanMessage("{user_message_key}")写法错误，HumanMessage是具体消息实例不支持模板变量替换，导致用户消息未被正确解析，整个上下文传递异常，过滤效果无法体现。

2. HumanMessage的正确用法 #

直接使用HumanMessage("{user_message_key}")是错误的，正确做法有两种：

• 在ChatPromptTemplate.from_messages中用("human", "{usr_msg}")的格式定义用户消息模板；
• 用HumanMessagePromptTemplate.from_template("{usr_msg}")显式创建模板。

input_messages_key的作用是告诉RunnableWithMessageHistory哪个输入变量是用户新消息，需要追加到历史中，因此必须与prompt中的变量名保持一致。

3. 在get_session_history中应用过滤的优缺点 #

优点 #

• 全局生效：所有获取历史的地方都会自动拿到过滤后的消息，无需在每个runnable链中重复编写逻辑；
• 节省资源：若历史消息量较大，过滤后可减少内存占用。

缺点 #

• 历史丢失：原始历史会被覆盖，后续无法恢复完整对话记录；
• 灵活性差：若不同场景需要不同过滤规则（如保留10条/5条），这种方式难以调整。

from typing import List, Union
from langchain_openai import ChatOpenAI
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables import RunnablePassthrough

store = {}
model_name = "gpt-3.5-turbo"
system_message = "You're a helpful, friendly AI"
model = ChatOpenAI(model=model_name)

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    history: BaseChatMessageHistory = store[session_id]
    # 可选：在这里过滤历史，会修改原始存储的历史
    # history.messages = filter_messages(history.messages)
    return history

def filter_messages(messages: List[Union[HumanMessage, AIMessage]]) -> List[Union[HumanMessage, AIMessage]]:
    return messages[-5:]

def process_text(text: str, session_id: str) -> AIMessage:
    # 正确构建prompt模板：用("human", "{usr_msg}")替代直接的HumanMessage
    prompt = ChatPromptTemplate.from_messages(
        [
            SystemMessage(system_message),
            MessagesPlaceholder(variable_name="chat_hist"),
            ("human", "{usr_msg}")
        ]
    )

    # 确保chat_hist是过滤后的历史
    runnable = RunnablePassthrough.assign(
        chat_hist=lambda x: filter_messages(x["chat_hist"])
    ) | prompt | model

    with_message_history = RunnableWithMessageHistory(
        runnable=runnable,
        get_session_history=get_session_history,
        history_messages_key="chat_hist",
        input_messages_key="usr_msg",  # 与输入变量名对应
    )

    # 直接传入字符串，input_messages_key会自动处理消息追加
    resp: AIMessage = with_message_history.invoke(
        {"usr_msg": text},
        config={"configurable": {"session_id": session_id}},
    )
    print(resp.content)
    return resp

session_id = "test_session"
previous_messages = [
    HumanMessage(content="I love the color Pink. It's my favorite color."),
    AIMessage(content="That's great to know!"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Here's a placeholder message for testing"),
    HumanMessage(content="Hello, my name is John."),
    AIMessage(content="Hi John, great to meet you."),
    HumanMessage(content="I'm sick of this rain."),
]
session_history = get_session_history(session_id)
for message in previous_messages:
    session_history.add_message(message)

user_input = "Repeat exactly: GOODBYE"
response: AIMessage = process_text(user_input, session_id)
# 正确返回：GOODBYE

• 将HumanMessage("{user_message_key}")改为("human", "{usr_msg}")，让prompt模板正确解析用户输入；
• 调用invoke时直接传入字符串{"usr_msg": text}，无需包装成HumanMessage列表，input_messages_key会自动处理消息追加；
• 确保RunnablePassthrough中的chat_hist过滤逻辑正确作用于注入的历史消息。

内容的提问来源于stack exchange，提问作者Peter