"""Module contains utility functions for working with messages.Some examples of what you can do with these functions include:* Convert messages to strings (serialization)* Convert messages from dicts to Message objects (deserialization)* Filter messages from a list of messages based on name, type or id etc."""from__future__importannotationsimportinspectimportjsonfromfunctoolsimportpartialfromtypingimport(TYPE_CHECKING,Any,Callable,Dict,Iterable,List,Literal,Optional,Sequence,Tuple,Type,Union,cast,overload,)fromlangchain_core.messages.aiimportAIMessage,AIMessageChunkfromlangchain_core.messages.baseimportBaseMessage,BaseMessageChunkfromlangchain_core.messages.chatimportChatMessage,ChatMessageChunkfromlangchain_core.messages.functionimportFunctionMessage,FunctionMessageChunkfromlangchain_core.messages.humanimportHumanMessage,HumanMessageChunkfromlangchain_core.messages.modifierimportRemoveMessagefromlangchain_core.messages.systemimportSystemMessage,SystemMessageChunkfromlangchain_core.messages.toolimportToolMessage,ToolMessageChunkifTYPE_CHECKING:fromlangchain_text_splittersimportTextSplitterfromlangchain_core.language_modelsimportBaseLanguageModelfromlangchain_core.prompt_valuesimportPromptValuefromlangchain_core.runnables.baseimportRunnableAnyMessage=Union[AIMessage,HumanMessage,ChatMessage,SystemMessage,FunctionMessage,ToolMessage,]
[docs]defget_buffer_string(messages:Sequence[BaseMessage],human_prefix:str="Human",ai_prefix:str="AI")->str:"""Convert a sequence of Messages to strings and concatenate them into one string. Args: messages: Messages to be converted to strings. human_prefix: The prefix to prepend to contents of HumanMessages. Default is "Human". ai_prefix: THe prefix to prepend to contents of AIMessages. Default is "AI". Returns: A single string concatenation of all input messages. Raises: ValueError: If an unsupported message type is encountered. Example: .. code-block:: python from langchain_core import AIMessage, HumanMessage messages = [ HumanMessage(content="Hi, how are you?"), AIMessage(content="Good, how are you?"), ] get_buffer_string(messages) # -> "Human: Hi, how are you?\nAI: Good, how are you?" """string_messages=[]forminmessages:ifisinstance(m,HumanMessage):role=human_prefixelifisinstance(m,AIMessage):role=ai_prefixelifisinstance(m,SystemMessage):role="System"elifisinstance(m,FunctionMessage):role="Function"elifisinstance(m,ToolMessage):role="Tool"elifisinstance(m,ChatMessage):role=m.roleelse:raiseValueError(f"Got unsupported message type: {m}")message=f"{role}: {m.content}"ifisinstance(m,AIMessage)and"function_call"inm.additional_kwargs:message+=f"{m.additional_kwargs['function_call']}"string_messages.append(message)return"\n".join(string_messages)
[docs]defmessages_from_dict(messages:Sequence[dict])->List[BaseMessage]:"""Convert a sequence of messages from dicts to Message objects. Args: messages: Sequence of messages (as dicts) to convert. Returns: List of messages (BaseMessages). """return[_message_from_dict(m)forminmessages]
[docs]defmessage_chunk_to_message(chunk:BaseMessageChunk)->BaseMessage:"""Convert a message chunk to a message. Args: chunk: Message chunk to convert. Returns: Message. """ifnotisinstance(chunk,BaseMessageChunk):returnchunk# chunk classes always have the equivalent non-chunk class as their first parentignore_keys=["type"]ifisinstance(chunk,AIMessageChunk):ignore_keys.append("tool_call_chunks")returnchunk.__class__.__mro__[1](**{k:vfork,vinchunk.__dict__.items()ifknotinignore_keys})
MessageLikeRepresentation=Union[BaseMessage,List[str],Tuple[str,str],str,Dict[str,Any]]def_create_message_from_message_type(message_type:str,content:str,name:Optional[str]=None,tool_call_id:Optional[str]=None,tool_calls:Optional[List[Dict[str,Any]]]=None,id:Optional[str]=None,**additional_kwargs:Any,)->BaseMessage:"""Create a message from a message type and content string. Args: message_type: (str) the type of the message (e.g., "human", "ai", etc.). content: (str) the content string. name: (str) the name of the message. Default is None. tool_call_id: (str) the tool call id. Default is None. tool_calls: (List[Dict[str, Any]]) the tool calls. Default is None. id: (str) the id of the message. Default is None. **additional_kwargs: (Dict[str, Any]) additional keyword arguments. Returns: a message of the appropriate type. Raises: ValueError: if the message type is not one of "human", "user", "ai", "assistant", "system", "function", or "tool". """kwargs:Dict[str,Any]={}ifnameisnotNone:kwargs["name"]=nameiftool_call_idisnotNone:kwargs["tool_call_id"]=tool_call_idifadditional_kwargs:kwargs["additional_kwargs"]=additional_kwargs# type: ignore[assignment]ifidisnotNone:kwargs["id"]=idiftool_callsisnotNone:kwargs["tool_calls"]=[]fortool_callintool_calls:# Convert OpenAI-format tool call to LangChain format.if"function"intool_call:args=tool_call["function"]["arguments"]ifisinstance(args,str):args=json.loads(args,strict=False)kwargs["tool_calls"].append({"name":tool_call["function"]["name"],"args":args,"id":tool_call["id"],"type":"tool_call",})else:kwargs["tool_calls"].append(tool_call)ifmessage_typein("human","user"):message:BaseMessage=HumanMessage(content=content,**kwargs)elifmessage_typein("ai","assistant"):message=AIMessage(content=content,**kwargs)elifmessage_type=="system":message=SystemMessage(content=content,**kwargs)elifmessage_type=="function":message=FunctionMessage(content=content,**kwargs)elifmessage_type=="tool":artifact=kwargs.get("additional_kwargs",{}).pop("artifact",None)message=ToolMessage(content=content,artifact=artifact,**kwargs)elifmessage_type=="remove":message=RemoveMessage(**kwargs)else:raiseValueError(f"Unexpected message type: {message_type}. Use one of 'human',"f" 'user', 'ai', 'assistant', 'function', 'tool', or 'system'.")returnmessagedef_convert_to_message(message:MessageLikeRepresentation)->BaseMessage:"""Instantiate a message from a variety of message formats. The message format can be one of the following: - BaseMessagePromptTemplate - BaseMessage - 2-tuple of (role string, template); e.g., ("human", "{user_input}") - dict: a message dict with role and content keys - string: shorthand for ("human", template); e.g., "{user_input}" Args: message: a representation of a message in one of the supported formats. Returns: an instance of a message or a message template. Raises: NotImplementedError: if the message type is not supported. ValueError: if the message dict does not contain the required keys. """ifisinstance(message,BaseMessage):_message=messageelifisinstance(message,str):_message=_create_message_from_message_type("human",message)elifisinstance(message,Sequence)andlen(message)==2:# mypy doesn't realise this can't be a string given the previous branchmessage_type_str,template=message# type: ignore[misc]_message=_create_message_from_message_type(message_type_str,template)elifisinstance(message,dict):msg_kwargs=message.copy()try:try:msg_type=msg_kwargs.pop("role")exceptKeyError:msg_type=msg_kwargs.pop("type")# None msg content is not allowedmsg_content=msg_kwargs.pop("content")or""exceptKeyErrorase:raiseValueError(f"Message dict must contain 'role' and 'content' keys, got {message}")frome_message=_create_message_from_message_type(msg_type,msg_content,**msg_kwargs)else:raiseNotImplementedError(f"Unsupported message type: {type(message)}")return_message
[docs]defconvert_to_messages(messages:Union[Iterable[MessageLikeRepresentation],PromptValue],)->List[BaseMessage]:"""Convert a sequence of messages to a list of messages. Args: messages: Sequence of messages to convert. Returns: List of messages (BaseMessages). """# Import here to avoid circular importsfromlangchain_core.prompt_valuesimportPromptValueifisinstance(messages,PromptValue):returnmessages.to_messages()return[_convert_to_message(m)forminmessages]
[docs]@_runnable_supportdeffilter_messages(messages:Union[Iterable[MessageLikeRepresentation],PromptValue],*,include_names:Optional[Sequence[str]]=None,exclude_names:Optional[Sequence[str]]=None,include_types:Optional[Sequence[Union[str,Type[BaseMessage]]]]=None,exclude_types:Optional[Sequence[Union[str,Type[BaseMessage]]]]=None,include_ids:Optional[Sequence[str]]=None,exclude_ids:Optional[Sequence[str]]=None,)->List[BaseMessage]:"""Filter messages based on name, type or id. Args: messages: Sequence Message-like objects to filter. include_names: Message names to include. Default is None. exclude_names: Messages names to exclude. Default is None. include_types: Message types to include. Can be specified as string names (e.g. "system", "human", "ai", ...) or as BaseMessage classes (e.g. SystemMessage, HumanMessage, AIMessage, ...). Default is None. exclude_types: Message types to exclude. Can be specified as string names (e.g. "system", "human", "ai", ...) or as BaseMessage classes (e.g. SystemMessage, HumanMessage, AIMessage, ...). Default is None. include_ids: Message IDs to include. Default is None. exclude_ids: Message IDs to exclude. Default is None. Returns: A list of Messages that meets at least one of the incl_* conditions and none of the excl_* conditions. If not incl_* conditions are specified then anything that is not explicitly excluded will be included. Raises: ValueError if two incompatible arguments are provided. Example: .. code-block:: python from langchain_core.messages import filter_messages, AIMessage, HumanMessage, SystemMessage messages = [ SystemMessage("you're a good assistant."), HumanMessage("what's your name", id="foo", name="example_user"), AIMessage("steve-o", id="bar", name="example_assistant"), HumanMessage("what's your favorite color", id="baz",), AIMessage("silicon blue", id="blah",), ] filter_messages( messages, incl_names=("example_user", "example_assistant"), incl_types=("system",), excl_ids=("bar",), ) .. code-block:: python [ SystemMessage("you're a good assistant."), HumanMessage("what's your name", id="foo", name="example_user"), ] """# noqa: E501messages=convert_to_messages(messages)filtered:List[BaseMessage]=[]formsginmessages:ifexclude_namesandmsg.nameinexclude_names:continueelifexclude_typesand_is_message_type(msg,exclude_types):continueelifexclude_idsandmsg.idinexclude_ids:continueelse:pass# default to inclusion when no inclusion criteria given.ifnot(include_typesorinclude_idsorinclude_names):filtered.append(msg)elifinclude_namesandmsg.nameininclude_names:filtered.append(msg)elifinclude_typesand_is_message_type(msg,include_types):filtered.append(msg)elifinclude_idsandmsg.idininclude_ids:filtered.append(msg)else:passreturnfiltered
[docs]@_runnable_supportdefmerge_message_runs(messages:Union[Iterable[MessageLikeRepresentation],PromptValue],*,chunk_separator:str="\n",)->List[BaseMessage]:"""Merge consecutive Messages of the same type. **NOTE**: ToolMessages are not merged, as each has a distinct tool call id that can't be merged. Args: messages: Sequence Message-like objects to merge. chunk_separator: Specify the string to be inserted between message chunks. Default is "\n". Returns: List of BaseMessages with consecutive runs of message types merged into single messages. By default, if two messages being merged both have string contents, the merged content is a concatenation of the two strings with a new-line separator. The separator inserted between message chunks can be controlled by specifying any string with ``chunk_separator``. If at least one of the messages has a list of content blocks, the merged content is a list of content blocks. Example: .. code-block:: python from langchain_core.messages import ( merge_message_runs, AIMessage, HumanMessage, SystemMessage, ToolCall, ) messages = [ SystemMessage("you're a good assistant."), HumanMessage("what's your favorite color", id="foo",), HumanMessage("wait your favorite food", id="bar",), AIMessage( "my favorite colo", tool_calls=[ToolCall(name="blah_tool", args={"x": 2}, id="123", type="tool_call")], id="baz", ), AIMessage( [{"type": "text", "text": "my favorite dish is lasagna"}], tool_calls=[ToolCall(name="blah_tool", args={"x": -10}, id="456", type="tool_call")], id="blur", ), ] merge_message_runs(messages) .. code-block:: python [ SystemMessage("you're a good assistant."), HumanMessage("what's your favorite color\\nwait your favorite food", id="foo",), AIMessage( [ "my favorite colo", {"type": "text", "text": "my favorite dish is lasagna"} ], tool_calls=[ ToolCall({"name": "blah_tool", "args": {"x": 2}, "id": "123", "type": "tool_call"}), ToolCall({"name": "blah_tool", "args": {"x": -10}, "id": "456", "type": "tool_call"}) ] id="baz" ), ] """# noqa: E501ifnotmessages:return[]messages=convert_to_messages(messages)merged:List[BaseMessage]=[]formsginmessages:curr=msg.copy(deep=True)last=merged.pop()ifmergedelseNoneifnotlast:merged.append(curr)elifisinstance(curr,ToolMessage)ornotisinstance(curr,last.__class__):merged.extend([last,curr])else:last_chunk=_msg_to_chunk(last)curr_chunk=_msg_to_chunk(curr)if(isinstance(last_chunk.content,str)andisinstance(curr_chunk.content,str)andlast_chunk.contentandcurr_chunk.content):last_chunk.content+=chunk_separatormerged.append(_chunk_to_msg(last_chunk+curr_chunk))returnmerged
# TODO: Update so validation errors (for token_counter, for example) are raised on# init not at runtime.
[docs]@_runnable_supportdeftrim_messages(messages:Union[Iterable[MessageLikeRepresentation],PromptValue],*,max_tokens:int,token_counter:Union[Callable[[List[BaseMessage]],int],Callable[[BaseMessage],int],BaseLanguageModel,],strategy:Literal["first","last"]="last",allow_partial:bool=False,end_on:Optional[Union[str,Type[BaseMessage],Sequence[Union[str,Type[BaseMessage]]]]]=None,start_on:Optional[Union[str,Type[BaseMessage],Sequence[Union[str,Type[BaseMessage]]]]]=None,include_system:bool=False,text_splitter:Optional[Union[Callable[[str],List[str]],TextSplitter]]=None,)->List[BaseMessage]:"""Trim messages to be below a token count. Args: messages: Sequence of Message-like objects to trim. max_tokens: Max token count of trimmed messages. 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. strategy: Strategy for trimming. - "first": Keep the first <= n_count tokens of the messages. - "last": Keep the last <= n_count tokens of the messages. Default is "last". allow_partial: Whether to split a message if only part of the message can be included. If ``strategy="last"`` then the last partial contents of a message are included. If ``strategy="first"`` then the first partial contents of a message are included. Default is False. end_on: The message type to end on. If specified then every message after the last occurrence of this type is ignored. If ``strategy=="last"`` then this is done before we attempt to get the last ``max_tokens``. If ``strategy=="first"`` then this is done after we get the first ``max_tokens``. Can be specified as string names (e.g. "system", "human", "ai", ...) or as BaseMessage classes (e.g. SystemMessage, HumanMessage, AIMessage, ...). Can be a single type or a list of types. Default is None. start_on: The message type to start on. Should only be specified if ``strategy="last"``. If specified then every message before the first occurrence of this type is ignored. This is done after we trim the initial messages to the last ``max_tokens``. Does not apply to a SystemMessage at index 0 if ``include_system=True``. Can be specified as string names (e.g. "system", "human", "ai", ...) or as BaseMessage classes (e.g. SystemMessage, HumanMessage, AIMessage, ...). Can be a single type or a list of types. Default is None. include_system: Whether to keep the SystemMessage if there is one at index 0. Should only be specified if ``strategy="last"``. Default is False. text_splitter: Function or ``langchain_text_splitters.TextSplitter`` for splitting the string contents of a message. Only used if ``allow_partial=True``. If ``strategy="last"`` then the last split tokens from a partial message will be included. if ``strategy=="first"`` then the first split tokens from a partial message will be included. Token splitter assumes that separators are kept, so that split contents can be directly concatenated to recreate the original text. Defaults to splitting on newlines. Returns: List of trimmed BaseMessages. Raises: ValueError: if two incompatible arguments are specified or an unrecognized ``strategy`` is specified. Example: .. code-block:: python 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 First 30 tokens, not allowing partial messages: .. code-block:: python trim_messages(messages, max_tokens=30, token_counter=dummy_token_counter, strategy="first") .. code-block:: python [ 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"), ] First 30 tokens, allowing partial messages: .. code-block:: python trim_messages( messages, max_tokens=30, token_counter=dummy_token_counter, strategy="first", allow_partial=True, ) .. code-block:: python [ 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"), ] First 30 tokens, allowing partial messages, have to end on HumanMessage: .. code-block:: python trim_messages( messages, max_tokens=30, token_counter=dummy_token_counter, strategy="first" allow_partial=True, end_on="human", ) .. code-block:: python [ 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"), ] Last 30 tokens, including system message, not allowing partial messages: .. code-block:: python trim_messages(messages, max_tokens=30, include_system=True, token_counter=dummy_token_counter, strategy="last") .. code-block:: python [ 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"), ] Last 40 tokens, including system message, allowing partial messages: .. code-block:: python trim_messages( messages, max_tokens=40, token_counter=dummy_token_counter, strategy="last", allow_partial=True, include_system=True ) .. code-block:: python [ 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"), ] Last 30 tokens, including system message, allowing partial messages, end on HumanMessage: .. code-block:: python trim_messages( messages, max_tokens=30, token_counter=dummy_token_counter, strategy="last", end_on="human", include_system=True, allow_partial=True, ) .. code-block:: python [ 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"), ] Last 40 tokens, including system message, allowing partial messages, start on HumanMessage: .. code-block:: python trim_messages( messages, max_tokens=40, token_counter=dummy_token_counter, strategy="last", include_system=True, allow_partial=True, start_on="human" ) .. code-block:: python [ 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"), ] """# noqa: E501ifstart_onandstrategy=="first":raiseValueErrorifinclude_systemandstrategy=="first":raiseValueErrormessages=convert_to_messages(messages)ifhasattr(token_counter,"get_num_tokens_from_messages"):list_token_counter=token_counter.get_num_tokens_from_messageselifcallable(token_counter):if(list(inspect.signature(token_counter).parameters.values())[0].annotationisBaseMessage):deflist_token_counter(messages:Sequence[BaseMessage])->int:returnsum(token_counter(msg)formsginmessages)# type: ignore[arg-type, misc]else:list_token_counter=token_counter# type: ignore[assignment]else:raiseValueError(f"'token_counter' expected to be a model that implements "f"'get_num_tokens_from_messages()' or a function. Received object of type "f"{type(token_counter)}.")try:fromlangchain_text_splittersimportTextSplitterexceptImportError:text_splitter_fn:Optional[Callable]=cast(Optional[Callable],text_splitter)else:ifisinstance(text_splitter,TextSplitter):text_splitter_fn=text_splitter.split_textelse:text_splitter_fn=text_splittertext_splitter_fn=text_splitter_fnor_default_text_splitterifstrategy=="first":return_first_max_tokens(messages,max_tokens=max_tokens,token_counter=list_token_counter,text_splitter=text_splitter_fn,partial_strategy="first"ifallow_partialelseNone,end_on=end_on,)elifstrategy=="last":return_last_max_tokens(messages,max_tokens=max_tokens,token_counter=list_token_counter,allow_partial=allow_partial,include_system=include_system,start_on=start_on,end_on=end_on,text_splitter=text_splitter_fn,)else:raiseValueError(f"Unrecognized {strategy=}. Supported strategies are 'last' and 'first'.")
def_first_max_tokens(messages:Sequence[BaseMessage],*,max_tokens:int,token_counter:Callable[[List[BaseMessage]],int],text_splitter:Callable[[str],List[str]],partial_strategy:Optional[Literal["first","last"]]=None,end_on:Optional[Union[str,Type[BaseMessage],Sequence[Union[str,Type[BaseMessage]]]]]=None,)->List[BaseMessage]:messages=list(messages)idx=0foriinrange(len(messages)):iftoken_counter(messages[:-i]ifielsemessages)<=max_tokens:idx=len(messages)-ibreakifidx<len(messages)-1andpartial_strategy:included_partial=Falseifisinstance(messages[idx].content,list):excluded=messages[idx].copy(deep=True)num_block=len(excluded.content)ifpartial_strategy=="last":excluded.content=list(reversed(excluded.content))for_inrange(1,num_block):excluded.content=excluded.content[:-1]iftoken_counter(messages[:idx]+[excluded])<=max_tokens:messages=messages[:idx]+[excluded]idx+=1included_partial=Truebreakifincluded_partialandpartial_strategy=="last":excluded.content=list(reversed(excluded.content))ifnotincluded_partial:excluded=messages[idx].copy(deep=True)ifisinstance(excluded.content,list)andany(isinstance(block,str)orblock["type"]=="text"forblockinmessages[idx].content):text_block=next(blockforblockinmessages[idx].contentifisinstance(block,str)orblock["type"]=="text")text=(text_block["text"]ifisinstance(text_block,dict)elsetext_block)elifisinstance(excluded.content,str):text=excluded.contentelse:text=Noneiftext:split_texts=text_splitter(text)num_splits=len(split_texts)ifpartial_strategy=="last":split_texts=list(reversed(split_texts))for_inrange(num_splits-1):split_texts.pop()excluded.content="".join(split_texts)iftoken_counter(messages[:idx]+[excluded])<=max_tokens:ifpartial_strategy=="last":excluded.content="".join(reversed(split_texts))messages=messages[:idx]+[excluded]idx+=1breakifend_on:whileidx>0andnot_is_message_type(messages[idx-1],end_on):idx-=1returnmessages[:idx]def_last_max_tokens(messages:Sequence[BaseMessage],*,max_tokens:int,token_counter:Callable[[List[BaseMessage]],int],text_splitter:Callable[[str],List[str]],allow_partial:bool=False,include_system:bool=False,start_on:Optional[Union[str,Type[BaseMessage],Sequence[Union[str,Type[BaseMessage]]]]]=None,end_on:Optional[Union[str,Type[BaseMessage],Sequence[Union[str,Type[BaseMessage]]]]]=None,)->List[BaseMessage]:messages=list(messages)ifend_on:whilemessagesandnot_is_message_type(messages[-1],end_on):messages.pop()swapped_system=include_systemandisinstance(messages[0],SystemMessage)ifswapped_system:reversed_=messages[:1]+messages[1:][::-1]else:reversed_=messages[::-1]reversed_=_first_max_tokens(reversed_,max_tokens=max_tokens,token_counter=token_counter,text_splitter=text_splitter,partial_strategy="last"ifallow_partialelseNone,end_on=start_on,)ifswapped_system:returnreversed_[:1]+reversed_[1:][::-1]else:returnreversed_[::-1]_MSG_CHUNK_MAP:Dict[Type[BaseMessage],Type[BaseMessageChunk]]={HumanMessage:HumanMessageChunk,AIMessage:AIMessageChunk,SystemMessage:SystemMessageChunk,ToolMessage:ToolMessageChunk,FunctionMessage:FunctionMessageChunk,ChatMessage:ChatMessageChunk,}_CHUNK_MSG_MAP={v:kfork,vin_MSG_CHUNK_MAP.items()}def_msg_to_chunk(message:BaseMessage)->BaseMessageChunk:ifmessage.__class__in_MSG_CHUNK_MAP:return_MSG_CHUNK_MAP[message.__class__](**message.dict(exclude={"type"}))formsg_cls,chunk_clsin_MSG_CHUNK_MAP.items():ifisinstance(message,msg_cls):returnchunk_cls(**message.dict(exclude={"type"}))raiseValueError(f"Unrecognized message class {message.__class__}. Supported classes are "f"{list(_MSG_CHUNK_MAP.keys())}")def_chunk_to_msg(chunk:BaseMessageChunk)->BaseMessage:ifchunk.__class__in_CHUNK_MSG_MAP:return_CHUNK_MSG_MAP[chunk.__class__](**chunk.dict(exclude={"type","tool_call_chunks"}))forchunk_cls,msg_clsin_CHUNK_MSG_MAP.items():ifisinstance(chunk,chunk_cls):returnmsg_cls(**chunk.dict(exclude={"type","tool_call_chunks"}))raiseValueError(f"Unrecognized message chunk class {chunk.__class__}. Supported classes are "f"{list(_CHUNK_MSG_MAP.keys())}")def_default_text_splitter(text:str)->List[str]:splits=text.split("\n")return[s+"\n"forsinsplits[:-1]]+splits[-1:]def_is_message_type(message:BaseMessage,type_:Union[str,Type[BaseMessage],Sequence[Union[str,Type[BaseMessage]]]],)->bool:types=[type_]ifisinstance(type_,(str,type))elsetype_types_str=[tfortintypesifisinstance(t,str)]types_types=tuple(tfortintypesifisinstance(t,type))returnmessage.typeintypes_strorisinstance(message,types_types)