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"), ]