langchain_core.messages.utils.trim_messages

langchain_core.messages.utils.trim_messages(messages: Optional[Sequence[MessageLikeRepresentation]] = None, **kwargs: Any) Union[List[BaseMessage], Runnable[Sequence[MessageLikeRepresentation], List[BaseMessage]]]

将消息修剪为低于令牌计数。

参数
  • messages (可选[序列[MessageLikeRepresentation]]) - 要修剪的类似于消息的对象的序列。

  • max_tokens - 修剪消息的最大令牌计数。

  • token_counter - BaseMessage 或 BaseMessage 列表中令牌计数的函数或 llm。如果传递了 BaseLanguageModel,则将使用 BaseLanguageModel.get_num_tokens_from_messages()。

  • strategy - 修剪策略。 - “first”:保留消息的前 n_count 个令牌。 - “last”:保留消息的最后 n_count 个令牌。默认为 "last"。

  • allow_partial - 是否在消息仅能包含部分内容时分割消息。如果 strategy="last",则包含消息的最后部分内容。如果 strategy="first",则包含消息的第一部分内容。默认为 False。

  • end_on - 要结束的消息类型。如果指定了,则忽略此类型之后的最后一个实例之后的每个消息。如果 strategy=="last",则在尝试获取最后一个 max_tokens 之前完成此操作。如果 strategy=="first",则在获取第一个 max_tokens 之后完成此操作。可以是字符串名称(例如,“system”,“human”,“ai”,……)或 BaseMessage 类(例如 SystemMessage,HumanMessage,AIMessage,……)。可以是单个类型或类型的列表。默认为 None。

  • start_on – 开始的消息类型。仅在 strategy="last" 时指定。如果指定,则忽略此类型首次出现之前的所有消息。在截断初始消息到最近的 max_tokens 之后执行。如果 include_system=True,则不适用于索引 0 的 SystemMessage。可以指定为字符串名称(例如“system”,“human”,“ai”,…)或基本消息类(例如 SystemMessage,HumanMessage,AIMessage,…)。可以是单个类型或类型列表。默认为 None。

  • include_system – 如果在索引 0 处有 SystemMessage,是否保留它。仅在 strategy="last" 时指定。默认为 False。

  • text_splitter – 用于拆分消息字符串内容的函数或 langchain_text_splitters.TextSplitter。仅在 allow_partial=True 时使用。如果 strategy="last",则包括部分消息的最后拆分标记。如果 strategy="first",则包括部分消息的第一个拆分标记。标记拆分器假设分隔符被保留,因此可以直接连接拆分内容以重新创建原始文本。默认为按换行符拆分。

  • kwargs (任意) –

返回

截断后的基本消息列表。

引发

ValueError – 如果指定了两个不可兼容的参数或指定了未识别的 strategy

返回类型

Union[List[BaseMessage], Runnable[Sequence[MessageLikeRepresentation], List[BaseMessage]]]

示例

from typing import List

from langchain_core.messages import trim_messages, AIMessage, BaseMessage, HumanMessage, SystemMessage

messages = [
    SystemMessage("This is a 4 token text. The full message is 10 tokens."),
    HumanMessage("This is a 4 token text. The full message is 10 tokens.", id="first"),
    AIMessage(
        [
            {"type": "text", "text": "This is the FIRST 4 token block."},
            {"type": "text", "text": "This is the SECOND 4 token block."},
        ],
        id="second",
    ),
    HumanMessage("This is a 4 token text. The full message is 10 tokens.", id="third"),
    AIMessage("This is a 4 token text. The full message is 10 tokens.", id="fourth"),
]

def dummy_token_counter(messages: List[BaseMessage]) -> int:
    # treat each message like it adds 3 default tokens at the beginning
    # of the message and at the end of the message. 3 + 4 + 3 = 10 tokens
    # per message.

    default_content_len = 4
    default_msg_prefix_len = 3
    default_msg_suffix_len = 3

    count = 0
    for msg in messages:
        if isinstance(msg.content, str):
            count += default_msg_prefix_len + default_content_len + default_msg_suffix_len
        if isinstance(msg.content, list):
            count += default_msg_prefix_len + len(msg.content) *  default_content_len + default_msg_suffix_len
    return count
前 30 个标记,不允许部分消息
trim_messages(messages, max_tokens=30, token_counter=dummy_token_counter, strategy="first")
[
    SystemMessage("This is a 4 token text. The full message is 10 tokens."),
    HumanMessage("This is a 4 token text. The full message is 10 tokens.", id="first"),
]
前 30 个标记,允许部分消息
trim_messages(
    messages,
    max_tokens=30,
    token_counter=dummy_token_counter,
    strategy="first",
    allow_partial=True,
)
[
    SystemMessage("This is a 4 token text. The full message is 10 tokens."),
    HumanMessage("This is a 4 token text. The full message is 10 tokens.", id="first"),
    AIMessage( [{"type": "text", "text": "This is the FIRST 4 token block."}], id="second"),
]
前 30 个标记,允许部分消息,必须在 HumanMessage 上结束
trim_messages(
    messages,
    max_tokens=30,
    token_counter=dummy_token_counter,
    strategy="first"
    allow_partial=True,
    end_on="human",
)
[
    SystemMessage("This is a 4 token text. The full message is 10 tokens."),
    HumanMessage("This is a 4 token text. The full message is 10 tokens.", id="first"),
]
包括系统消息的最后 30 个标记,不允许部分消息
trim_messages(messages, max_tokens=30, include_system=True, token_counter=dummy_token_counter, strategy="last")
[
    SystemMessage("This is a 4 token text. The full message is 10 tokens."),
    HumanMessage("This is a 4 token text. The full message is 10 tokens.", id="third"),
    AIMessage("This is a 4 token text. The full message is 10 tokens.", id="fourth"),
]
包括系统消息的最后 40 个标记,允许部分消息
trim_messages(
    messages,
    max_tokens=40,
    token_counter=dummy_token_counter,
    strategy="last",
    allow_partial=True,
    include_system=True
)
[
    SystemMessage("This is a 4 token text. The full message is 10 tokens."),
    AIMessage(
        [{"type": "text", "text": "This is the FIRST 4 token block."},],
        id="second",
    ),
    HumanMessage("This is a 4 token text. The full message is 10 tokens.", id="third"),
    AIMessage("This is a 4 token text. The full message is 10 tokens.", id="fourth"),
]
包括系统消息的最后 30 个标记,允许部分消息,必须在 HumanMessage 上结束
trim_messages(
    messages,
    max_tokens=30,
    token_counter=dummy_token_counter,
    strategy="last",
    end_on="human",
    include_system=True,
    allow_partial=True,
)
[
    SystemMessage("This is a 4 token text. The full message is 10 tokens."),
    AIMessage(
        [{"type": "text", "text": "This is the FIRST 4 token block."},],
        id="second",
    ),
    HumanMessage("This is a 4 token text. The full message is 10 tokens.", id="third"),
]
包括系统消息的最后 40 个标记,允许部分消息,从 HumanMessage 开始
trim_messages(
    messages,
    max_tokens=40,
    token_counter=dummy_token_counter,
    strategy="last",
    include_system=True,
    allow_partial=True,
    start_on="human"
)
[
    SystemMessage("This is a 4 token text. The full message is 10 tokens."),
    HumanMessage("This is a 4 token text. The full message is 10 tokens.", id="third"),
    AIMessage("This is a 4 token text. The full message is 10 tokens.", id="fourth"),
]