codekingpro/portable-devtools
114k
1"""Module contains utility functions for working with messages.2 3Some examples of what you can do with these functions include:4 5* Convert messages to strings (serialization)6* Convert messages from dicts to Message objects (deserialization)7* Filter messages from a list of messages based on name, type or id etc.8"""9 10from __future__ import annotations11 12import base6413import inspect14import json15import logging16import math17from collections.abc import Callable, Iterable, Sequence18from functools import partial, wraps19from typing import (20 TYPE_CHECKING,21 Annotated,22 Any,23 Concatenate,24 Literal,25 ParamSpec,26 Protocol,27 TypeVar,28 cast,29 overload,30)31from xml.sax.saxutils import escape, quoteattr32 33from pydantic import Discriminator, Field, Tag34 35from langchain_core.exceptions import ErrorCode, create_message36from langchain_core.messages.ai import AIMessage, AIMessageChunk37from langchain_core.messages.base import BaseMessage, BaseMessageChunk38from langchain_core.messages.block_translators.openai import (39 convert_to_openai_data_block,40)41from langchain_core.messages.chat import ChatMessage, ChatMessageChunk42from langchain_core.messages.content import (43 is_data_content_block,44)45from langchain_core.messages.function import FunctionMessage, FunctionMessageChunk46from langchain_core.messages.human import HumanMessage, HumanMessageChunk47from langchain_core.messages.modifier import RemoveMessage48from langchain_core.messages.system import SystemMessage, SystemMessageChunk49from langchain_core.messages.tool import ToolCall, ToolMessage, ToolMessageChunk50from langchain_core.utils.function_calling import convert_to_openai_tool51 52if TYPE_CHECKING:53 from langchain_core.language_models import BaseLanguageModel54 from langchain_core.prompt_values import PromptValue55 from langchain_core.runnables.base import Runnable56 from langchain_core.tools import BaseTool57 58try:59 from langchain_text_splitters import TextSplitter60 61 _HAS_LANGCHAIN_TEXT_SPLITTERS = True62except ImportError:63 _HAS_LANGCHAIN_TEXT_SPLITTERS = False64 65logger = logging.getLogger(__name__)66 67 68def _get_type(v: Any) -> str:69 """Get the type associated with the object for serialization purposes."""70 if isinstance(v, dict) and "type" in v:71 result = v["type"]72 elif hasattr(v, "type"):73 result = v.type74 else:75 msg = (76 f"Expected either a dictionary with a 'type' key or an object "77 f"with a 'type' attribute. Instead got type {type(v)}."78 )79 raise TypeError(msg)80 if not isinstance(result, str):81 msg = f"Expected 'type' to be a str, got {type(result).__name__}"82 raise TypeError(msg)83 return result84 85 86AnyMessage = Annotated[87 Annotated[AIMessage, Tag(tag="ai")]88 | Annotated[HumanMessage, Tag(tag="human")]89 | Annotated[ChatMessage, Tag(tag="chat")]90 | Annotated[SystemMessage, Tag(tag="system")]91 | Annotated[FunctionMessage, Tag(tag="function")]92 | Annotated[ToolMessage, Tag(tag="tool")]93 | Annotated[AIMessageChunk, Tag(tag="AIMessageChunk")]94 | Annotated[HumanMessageChunk, Tag(tag="HumanMessageChunk")]95 | Annotated[ChatMessageChunk, Tag(tag="ChatMessageChunk")]96 | Annotated[SystemMessageChunk, Tag(tag="SystemMessageChunk")]97 | Annotated[FunctionMessageChunk, Tag(tag="FunctionMessageChunk")]98 | Annotated[ToolMessageChunk, Tag(tag="ToolMessageChunk")],99 Field(discriminator=Discriminator(_get_type)),100]101"""A type representing any defined `Message` or `MessageChunk` type."""102 103 104def _has_base64_data(block: dict) -> bool:105 """Check if a content block contains base64 encoded data.106 107 Args:108 block: A content block dictionary.109 110 Returns:111 Whether the block contains base64 data.112 """113 # Check for explicit base64 field (standard content blocks)114 if block.get("base64"):115 return True116 117 # Check for data: URL in url field118 url = block.get("url", "")119 if isinstance(url, str) and url.startswith("data:"):120 return True121 122 # Check for OpenAI-style image_url with data: URL123 image_url = block.get("image_url", {})124 if isinstance(image_url, dict):125 url = image_url.get("url", "")126 if isinstance(url, str) and url.startswith("data:"):127 return True128 129 return False130 131 132_XML_CONTENT_BLOCK_MAX_LEN = 500133 134 135def _truncate(text: str, max_len: int = _XML_CONTENT_BLOCK_MAX_LEN) -> str:136 """Truncate text to `max_len` characters, adding ellipsis if truncated."""137 if len(text) <= max_len:138 return text139 return text[:max_len] + "..."140 141 142def _format_content_block_xml(block: dict) -> str | None:143 """Format a content block as XML.144 145 Args:146 block: A LangChain content block.147 148 Returns:149 XML string representation of the block, or `None` if the block should be150 skipped.151 152 Note:153 Plain text document content, server tool call arguments, and server tool154 result outputs are truncated to 500 characters.155 """156 block_type = block.get("type", "")157 158 # Skip blocks with base64 encoded data159 if _has_base64_data(block):160 return None161 162 # Text blocks163 if block_type == "text":164 text = block.get("text", "")165 return escape(text) if text else None166 167 # Reasoning blocks168 if block_type == "reasoning":169 reasoning = block.get("reasoning", "")170 if reasoning:171 return f"<reasoning>{escape(reasoning)}</reasoning>"172 return None173 174 # Image blocks (URL only, base64 already filtered)175 if block_type == "image":176 url = block.get("url")177 file_id = block.get("file_id")178 if url:179 return f"<image url={quoteattr(url)} />"180 if file_id:181 return f"<image file_id={quoteattr(file_id)} />"182 return None183 184 # OpenAI-style image_url blocks185 if block_type == "image_url":186 image_url = block.get("image_url", {})187 if isinstance(image_url, dict):188 url = image_url.get("url", "")189 if url and not url.startswith("data:"):190 return f"<image url={quoteattr(url)} />"191 return None192 193 # Audio blocks (URL only)194 if block_type == "audio":195 url = block.get("url")196 file_id = block.get("file_id")197 if url:198 return f"<audio url={quoteattr(url)} />"199 if file_id:200 return f"<audio file_id={quoteattr(file_id)} />"201 return None202 203 # Video blocks (URL only)204 if block_type == "video":205 url = block.get("url")206 file_id = block.get("file_id")207 if url:208 return f"<video url={quoteattr(url)} />"209 if file_id:210 return f"<video file_id={quoteattr(file_id)} />"211 return None212 213 # Plain text document blocks214 if block_type == "text-plain":215 text = block.get("text", "")216 return escape(_truncate(text)) if text else None217 218 # Server tool call blocks (from AI messages)219 if block_type == "server_tool_call":220 tc_id = quoteattr(str(block.get("id") or ""))221 tc_name = quoteattr(str(block.get("name") or ""))222 tc_args_json = json.dumps(block.get("args", {}), ensure_ascii=False)223 tc_args = escape(_truncate(tc_args_json))224 return (225 f"<server_tool_call id={tc_id} name={tc_name}>{tc_args}</server_tool_call>"226 )227 228 # Server tool result blocks229 if block_type == "server_tool_result":230 tool_call_id = quoteattr(str(block.get("tool_call_id") or ""))231 status = quoteattr(str(block.get("status") or ""))232 output = block.get("output")233 if output:234 output_json = json.dumps(output, ensure_ascii=False)235 output_str = escape(_truncate(output_json))236 else:237 output_str = ""238 return (239 f"<server_tool_result tool_call_id={tool_call_id} status={status}>"240 f"{output_str}</server_tool_result>"241 )242 243 # Unknown block type - skip silently244 return None245 246 247def _get_message_type_str(248 m: BaseMessage,249 human_prefix: str,250 ai_prefix: str,251 system_prefix: str,252 function_prefix: str,253 tool_prefix: str,254) -> str:255 """Get the type string for XML message element.256 257 Args:258 m: The message to get the type string for.259 human_prefix: The prefix to use for `HumanMessage`.260 ai_prefix: The prefix to use for `AIMessage`.261 system_prefix: The prefix to use for `SystemMessage`.262 function_prefix: The prefix to use for `FunctionMessage`.263 tool_prefix: The prefix to use for `ToolMessage`.264 265 Returns:266 The type string for the message element.267 268 Raises:269 ValueError: If an unsupported message type is encountered.270 """271 if isinstance(m, HumanMessage):272 return human_prefix.lower()273 if isinstance(m, AIMessage):274 return ai_prefix.lower()275 if isinstance(m, SystemMessage):276 return system_prefix.lower()277 if isinstance(m, FunctionMessage):278 return function_prefix.lower()279 if isinstance(m, ToolMessage):280 return tool_prefix.lower()281 if isinstance(m, ChatMessage):282 return m.role283 msg = f"Got unsupported message type: {m}"284 raise ValueError(msg)285 286 287def get_buffer_string(288 messages: Sequence[BaseMessage],289 human_prefix: str = "Human",290 ai_prefix: str = "AI",291 *,292 system_prefix: str = "System",293 function_prefix: str = "Function",294 tool_prefix: str = "Tool",295 message_separator: str = "\n",296 format: Literal["prefix", "xml"] = "prefix", # noqa: A002297) -> str:298 r"""Convert a sequence of messages to strings and concatenate them into one string.299 300 Args:301 messages: Messages to be converted to strings.302 human_prefix: The prefix to prepend to contents of `HumanMessage`s.303 ai_prefix: The prefix to prepend to contents of `AIMessage`.304 system_prefix: The prefix to prepend to contents of `SystemMessage`s.305 function_prefix: The prefix to prepend to contents of `FunctionMessage`s.306 tool_prefix: The prefix to prepend to contents of `ToolMessage`s.307 message_separator: The separator to use between messages.308 format: The output format. `'prefix'` uses `Role: content` format (default).309 310 `'xml'` uses XML-style `<message type='role'>` format with proper character311 escaping, which is useful when message content may contain role-like312 prefixes that could cause ambiguity.313 314 Returns:315 A single string concatenation of all input messages.316 317 Raises:318 ValueError: If an unsupported message type is encountered.319 320 !!! warning321 322 If a message is an `AIMessage` and contains both tool calls under `tool_calls`323 and a function call under `additional_kwargs["function_call"]`, only the tool324 calls will be appended to the string representation.325 326 !!! note "XML format"327 328 When using `format='xml'`:329 330 - All messages use uniform `<message type="role">content</message>` format.331 - The `type` attribute uses `human_prefix` (lowercased) for `HumanMessage`,332 `ai_prefix` (lowercased) for `AIMessage`, `system_prefix` (lowercased)333 for `SystemMessage`, `function_prefix` (lowercased) for `FunctionMessage`,334 `tool_prefix` (lowercased) for `ToolMessage`, and the original role335 (unchanged) for `ChatMessage`.336 - Message content is escaped using `xml.sax.saxutils.escape()`.337 - Attribute values are escaped using `xml.sax.saxutils.quoteattr()`.338 - AI messages with tool calls use nested structure with `<content>` and339 `<tool_call>` elements.340 - For multi-modal content (list of content blocks), supported block types341 are: `text`, `reasoning`, `image` (URL/file_id only), `image_url`342 (OpenAI-style, URL only), `audio` (URL/file_id only), `video` (URL/file_id343 only), `text-plain`, `server_tool_call`, and `server_tool_result`.344 - Content blocks with base64-encoded data are skipped (including blocks345 with `base64` field or `data:` URLs).346 - Unknown block types are skipped.347 - Plain text document content (`text-plain`), server tool call arguments,348 and server tool result outputs are truncated to 500 characters.349 350 Example:351 Default prefix format:352 353 ```python354 from langchain_core.messages import AIMessage, HumanMessage, get_buffer_string355 356 messages = [357 HumanMessage(content="Hi, how are you?"),358 AIMessage(content="Good, how are you?"),359 ]360 get_buffer_string(messages)361 # -> "Human: Hi, how are you?\nAI: Good, how are you?"362 ```363 364 XML format (useful when content contains role-like prefixes):365 366 ```python367 messages = [368 HumanMessage(content="Example: Human: some text"),369 AIMessage(content="I see the example."),370 ]371 get_buffer_string(messages, format="xml")372 # -> '<message type="human">Example: Human: some text</message>\\n'373 # -> '<message type="ai">I see the example.</message>'374 ```375 376 XML format with special characters (automatically escaped):377 378 ```python379 messages = [380 HumanMessage(content="Is 5 < 10 & 10 > 5?"),381 ]382 get_buffer_string(messages, format="xml")383 # -> '<message type="human">Is 5 < 10 & 10 > 5?</message>'384 ```385 386 XML format with tool calls:387 388 ```python389 messages = [390 AIMessage(391 content="I'll search for that.",392 tool_calls=[393 {"id": "call_123", "name": "search", "args": {"query": "weather"}}394 ],395 ),396 ]397 get_buffer_string(messages, format="xml")398 # -> '<message type="ai">\\n'399 # -> ' <content>I\\'ll search for that.</content>\\n'400 # -> ' <tool_call id="call_123" name="search">'401 # -> '{"query": "weather"}</tool_call>\\n'402 # -> '</message>'403 ```404 """405 if format not in {"prefix", "xml"}:406 msg = (407 f"Unrecognized format={format!r}. Supported formats are 'prefix' and 'xml'."408 )409 raise ValueError(msg)410 411 string_messages = []412 for m in messages:413 if isinstance(m, HumanMessage):414 role = human_prefix415 elif isinstance(m, AIMessage):416 role = ai_prefix417 elif isinstance(m, SystemMessage):418 role = system_prefix419 elif isinstance(m, FunctionMessage):420 role = function_prefix421 elif isinstance(m, ToolMessage):422 role = tool_prefix423 elif isinstance(m, ChatMessage):424 role = m.role425 else:426 msg = f"Got unsupported message type: {m}"427 raise ValueError(msg) # noqa: TRY004428 429 if format == "xml":430 msg_type = _get_message_type_str(431 m, human_prefix, ai_prefix, system_prefix, function_prefix, tool_prefix432 )433 434 # Format content blocks435 if isinstance(m.content, str):436 content_parts = [escape(m.content)] if m.content else []437 else:438 # List of content blocks439 content_parts = []440 for block in m.content:441 if isinstance(block, str):442 if block:443 content_parts.append(escape(block))444 else:445 formatted = _format_content_block_xml(block)446 if formatted:447 content_parts.append(formatted)448 449 # Check if this is an AIMessage with tool calls450 has_tool_calls = isinstance(m, AIMessage) and m.tool_calls451 has_function_call = (452 isinstance(m, AIMessage)453 and not m.tool_calls454 and "function_call" in m.additional_kwargs455 )456 457 if has_tool_calls or has_function_call:458 # Use nested structure for AI messages with tool calls459 # Type narrowing: at this point m is AIMessage (verified above)460 ai_msg = cast("AIMessage", m)461 parts = [f"<message type={quoteattr(msg_type)}>"]462 if content_parts:463 parts.append(f" <content>{' '.join(content_parts)}</content>")464 465 if has_tool_calls:466 for tc in ai_msg.tool_calls:467 tc_id = quoteattr(str(tc.get("id") or ""))468 tc_name = quoteattr(str(tc.get("name") or ""))469 tc_args = escape(470 json.dumps(tc.get("args", {}), ensure_ascii=False)471 )472 parts.append(473 f" <tool_call id={tc_id} name={tc_name}>"474 f"{tc_args}</tool_call>"475 )476 elif has_function_call:477 fc = ai_msg.additional_kwargs["function_call"]478 fc_name = quoteattr(str(fc.get("name") or ""))479 fc_args = escape(str(fc.get("arguments") or "{}"))480 parts.append(481 f" <function_call name={fc_name}>{fc_args}</function_call>"482 )483 484 parts.append("</message>")485 message = "\n".join(parts)486 else:487 # Simple structure for messages without tool calls488 joined_content = " ".join(content_parts)489 message = (490 f"<message type={quoteattr(msg_type)}>{joined_content}</message>"491 )492 else: # format == "prefix"493 content = m.text494 message = f"{role}: {content}"495 tool_info = ""496 if isinstance(m, AIMessage):497 if m.tool_calls:498 tool_info = str(m.tool_calls)499 elif "function_call" in m.additional_kwargs:500 # Legacy behavior assumes only one function call per message501 tool_info = str(m.additional_kwargs["function_call"])502 if tool_info:503 message += tool_info # Preserve original behavior504 505 string_messages.append(message)506 507 return message_separator.join(string_messages)508 509 510def _message_from_dict(message: dict) -> BaseMessage:511 type_ = message["type"]512 if type_ == "human":513 return HumanMessage(**message["data"])514 if type_ == "ai":515 return AIMessage(**message["data"])516 if type_ == "system":517 return SystemMessage(**message["data"])518 if type_ == "chat":519 return ChatMessage(**message["data"])520 if type_ == "function":521 return FunctionMessage(**message["data"])522 if type_ == "tool":523 return ToolMessage(**message["data"])524 if type_ == "remove":525 return RemoveMessage(**message["data"])526 if type_ == "AIMessageChunk":527 return AIMessageChunk(**message["data"])528 if type_ == "HumanMessageChunk":529 return HumanMessageChunk(**message["data"])530 if type_ == "FunctionMessageChunk":531 return FunctionMessageChunk(**message["data"])532 if type_ == "ToolMessageChunk":533 return ToolMessageChunk(**message["data"])534 if type_ == "SystemMessageChunk":535 return SystemMessageChunk(**message["data"])536 if type_ == "ChatMessageChunk":537 return ChatMessageChunk(**message["data"])538 msg = f"Got unexpected message type: {type_}"539 raise ValueError(msg)540 541 542def messages_from_dict(messages: Sequence[dict]) -> list[BaseMessage]:543 """Convert a sequence of messages from dicts to `Message` objects.544 545 Args:546 messages: Sequence of messages (as dicts) to convert.547 548 Returns:549 list of messages (BaseMessages).550 551 """552 return [_message_from_dict(m) for m in messages]553 554 555def message_chunk_to_message(chunk: BaseMessage) -> BaseMessage:556 """Convert a message chunk to a `Message`.557 558 Args:559 chunk: Message chunk to convert.560 561 Returns:562 Message.563 """564 if not isinstance(chunk, BaseMessageChunk):565 return chunk566 # chunk classes always have the equivalent non-chunk class as their first parent567 ignore_keys = ["type"]568 if isinstance(chunk, AIMessageChunk):569 ignore_keys.extend(["tool_call_chunks", "chunk_position"])570 return cast(571 "BaseMessage",572 chunk.__class__.__mro__[1](573 **{k: v for k, v in chunk.__dict__.items() if k not in ignore_keys}574 ),575 )576 577 578MessageLikeRepresentation = (579 BaseMessage | list[str] | tuple[str, str] | str | dict[str, Any]580)581"""A type representing the various ways a message can be represented."""582 583 584def _create_message_from_message_type(585 message_type: str,586 content: str,587 name: str | None = None,588 tool_call_id: str | None = None,589 tool_calls: list[dict[str, Any]] | None = None,590 id: str | None = None,591 **additional_kwargs: Any,592) -> BaseMessage:593 """Create a message from a `Message` type and content string.594 595 Args:596 message_type: the type of the message (e.g., `'human'`, `'ai'`, etc.).597 content: the content string.598 name: the name of the message.599 tool_call_id: the tool call id.600 tool_calls: the tool calls.601 id: the id of the message.602 additional_kwargs: additional keyword arguments.603 604 Returns:605 a message of the appropriate type.606 607 Raises:608 ValueError: if the message type is not one of `'human'`, `'user'`, `'ai'`,609 `'assistant'`, `'function'`, `'tool'`, `'system'`, or610 `'developer'`.611 """612 kwargs: dict[str, Any] = {}613 if name is not None:614 kwargs["name"] = name615 if tool_call_id is not None:616 kwargs["tool_call_id"] = tool_call_id617 if additional_kwargs:618 if response_metadata := additional_kwargs.pop("response_metadata", None):619 kwargs["response_metadata"] = response_metadata620 kwargs["additional_kwargs"] = additional_kwargs621 additional_kwargs.update(additional_kwargs.pop("additional_kwargs", {}))622 if id is not None:623 kwargs["id"] = id624 if tool_calls is not None:625 kwargs["tool_calls"] = []626 for tool_call in tool_calls:627 # Convert OpenAI-format tool call to LangChain format.628 if "function" in tool_call:629 args = tool_call["function"]["arguments"]630 if isinstance(args, str):631 args = json.loads(args, strict=False)632 kwargs["tool_calls"].append(633 {634 "name": tool_call["function"]["name"],635 "args": args,636 "id": tool_call["id"],637 "type": "tool_call",638 }639 )640 else:641 kwargs["tool_calls"].append(tool_call)642 if message_type in {"human", "user"}:643 if example := kwargs.get("additional_kwargs", {}).pop("example", False):644 kwargs["example"] = example645 message: BaseMessage = HumanMessage(content=content, **kwargs)646 elif message_type in {"ai", "assistant"}:647 if example := kwargs.get("additional_kwargs", {}).pop("example", False):648 kwargs["example"] = example649 message = AIMessage(content=content, **kwargs)650 elif message_type in {"system", "developer"}:651 if message_type == "developer":652 kwargs["additional_kwargs"] = kwargs.get("additional_kwargs") or {}653 kwargs["additional_kwargs"]["__openai_role__"] = "developer"654 message = SystemMessage(content=content, **kwargs)655 elif message_type == "function":656 message = FunctionMessage(content=content, **kwargs)657 elif message_type == "tool":658 artifact = kwargs.get("additional_kwargs", {}).pop("artifact", None)659 status = kwargs.get("additional_kwargs", {}).pop("status", None)660 if status is not None:661 kwargs["status"] = status662 message = ToolMessage(content=content, artifact=artifact, **kwargs)663 elif message_type == "remove":664 message = RemoveMessage(**kwargs)665 else:666 msg = (667 f"Unexpected message type: '{message_type}'. Use one of 'human',"668 f" 'user', 'ai', 'assistant', 'function', 'tool', 'system', or 'developer'."669 )670 msg = create_message(message=msg, error_code=ErrorCode.MESSAGE_COERCION_FAILURE)671 raise ValueError(msg)672 return message673 674 675def _convert_to_message(message: MessageLikeRepresentation) -> BaseMessage:676 """Instantiate a `Message` from a variety of message formats.677 678 The message format can be one of the following:679 680 - `BaseMessagePromptTemplate`681 - `BaseMessage`682 - 2-tuple of (role string, template); e.g., (`'human'`, `'{user_input}'`)683 - dict: a message dict with role and content keys684 - string: shorthand for (`'human'`, template); e.g., `'{user_input}'`685 686 Args:687 message: a representation of a message in one of the supported formats.688 689 Returns:690 An instance of a message or a message template.691 692 Raises:693 NotImplementedError: if the message type is not supported.694 ValueError: if the message dict does not contain the required keys.695 696 """697 if isinstance(message, BaseMessage):698 message_ = message699 elif isinstance(message, Sequence):700 if isinstance(message, str):701 message_ = _create_message_from_message_type("human", message)702 else:703 try:704 message_type_str, template = message705 except ValueError as e:706 msg = "Message as a sequence must be (role string, template)"707 raise NotImplementedError(msg) from e708 message_ = _create_message_from_message_type(message_type_str, template)709 elif isinstance(message, dict):710 msg_kwargs = message.copy()711 try:712 try:713 msg_type = msg_kwargs.pop("role")714 except KeyError:715 msg_type = msg_kwargs.pop("type")716 # None msg content is not allowed717 msg_content = msg_kwargs.pop("content") or ""718 except KeyError as e:719 msg = f"Message dict must contain 'role' and 'content' keys, got {message}"720 msg = create_message(721 message=msg, error_code=ErrorCode.MESSAGE_COERCION_FAILURE722 )723 raise ValueError(msg) from e724 message_ = _create_message_from_message_type(725 msg_type, msg_content, **msg_kwargs726 )727 else:728 msg = f"Unsupported message type: {type(message)}"729 msg = create_message(message=msg, error_code=ErrorCode.MESSAGE_COERCION_FAILURE)730 raise NotImplementedError(msg)731 732 return message_733 734 735def convert_to_messages(736 messages: Iterable[MessageLikeRepresentation] | PromptValue,737) -> list[BaseMessage]:738 """Convert a sequence of messages to a list of messages.739 740 Args:741 messages: Sequence of messages to convert.742 743 Returns:744 list of messages (BaseMessages).745 746 """747 # Import here to avoid circular imports748 from langchain_core.prompt_values import PromptValue # noqa: PLC0415749 750 if isinstance(messages, PromptValue):751 return messages.to_messages()752 return [_convert_to_message(m) for m in messages]753 754 755_P = ParamSpec("_P")756_R_co = TypeVar("_R_co", covariant=True)757 758 759class _RunnableSupportCallable(Protocol[_P, _R_co]):760 @overload761 def __call__(762 self,763 messages: None = None,764 *args: _P.args,765 **kwargs: _P.kwargs,766 ) -> Runnable[Sequence[MessageLikeRepresentation], _R_co]: ...767 768 @overload769 def __call__(770 self,771 messages: Sequence[MessageLikeRepresentation] | PromptValue,772 *args: _P.args,773 **kwargs: _P.kwargs,774 ) -> _R_co: ...775 776 def __call__(777 self,778 messages: Sequence[MessageLikeRepresentation] | PromptValue | None = None,779 *args: _P.args,780 **kwargs: _P.kwargs,781 ) -> _R_co | Runnable[Sequence[MessageLikeRepresentation], _R_co]: ...782 783 784def _runnable_support(785 func: Callable[786 Concatenate[Sequence[MessageLikeRepresentation] | PromptValue, _P], _R_co787 ],788) -> _RunnableSupportCallable[_P, _R_co]:789 @wraps(func)790 def wrapped(791 messages: Sequence[MessageLikeRepresentation] | PromptValue | None = None,792 *args: _P.args,793 **kwargs: _P.kwargs,794 ) -> _R_co | Runnable[Sequence[MessageLikeRepresentation], _R_co]:795 # Import locally to prevent circular import.796 from langchain_core.runnables.base import RunnableLambda # noqa: PLC0415797 798 if messages is not None:799 return func(messages, *args, **kwargs)800 return RunnableLambda(partial(func, **kwargs), name=func.__name__)801 802 return cast("_RunnableSupportCallable[_P, _R_co]", wrapped)803 804 805@_runnable_support806def filter_messages(807 messages: Iterable[MessageLikeRepresentation] | PromptValue,808 *,809 include_names: Sequence[str] | None = None,810 exclude_names: Sequence[str] | None = None,811 include_types: Sequence[str | type[BaseMessage]] | None = None,812 exclude_types: Sequence[str | type[BaseMessage]] | None = None,813 include_ids: Sequence[str] | None = None,814 exclude_ids: Sequence[str] | None = None,815 exclude_tool_calls: Sequence[str] | bool | None = None,816) -> list[BaseMessage]:817 """Filter messages based on `name`, `type` or `id`.818 819 Args:820 messages: Sequence Message-like objects to filter.821 include_names: Message names to include.822 exclude_names: Messages names to exclude.823 include_types: Message types to include. Can be specified as string names824 (e.g. `'system'`, `'human'`, `'ai'`, ...) or as `BaseMessage`825 classes (e.g. `SystemMessage`, `HumanMessage`, `AIMessage`, ...).826 827 exclude_types: Message types to exclude. Can be specified as string names828 (e.g. `'system'`, `'human'`, `'ai'`, ...) or as `BaseMessage`829 classes (e.g. `SystemMessage`, `HumanMessage`, `AIMessage`, ...).830 831 include_ids: Message IDs to include.832 exclude_ids: Message IDs to exclude.833 exclude_tool_calls: Tool call IDs to exclude.834 Can be one of the following:835 - `True`: All `AIMessage` objects with tool calls and all `ToolMessage`836 objects will be excluded.837 - a sequence of tool call IDs to exclude:838 - `ToolMessage` objects with the corresponding tool call ID will be839 excluded.840 - The `tool_calls` in the AIMessage will be updated to exclude841 matching tool calls. If all `tool_calls` are filtered from an842 AIMessage, the whole message is excluded.843 844 Returns:845 A list of Messages that meets at least one of the `incl_*` conditions and none846 of the `excl_*` conditions. If not `incl_*` conditions are specified then847 anything that is not explicitly excluded will be included.848 849 Raises:850 ValueError: If two incompatible arguments are provided.851 852 Example:853 ```python854 from langchain_core.messages import (855 filter_messages,856 AIMessage,857 HumanMessage,858 SystemMessage,859 )860 861 messages = [862 SystemMessage("you're a good assistant."),863 HumanMessage("what's your name", id="foo", name="example_user"),864 AIMessage("steve-o", id="bar", name="example_assistant"),865 HumanMessage(866 "what's your favorite color",867 id="baz",868 ),869 AIMessage(870 "silicon blue",871 id="blah",872 ),873 ]874 875 filter_messages(876 messages,877 include_names=("example_user", "example_assistant"),878 include_types=("system",),879 exclude_ids=("bar",),880 )881 ```882 883 ```python884 [885 SystemMessage("you're a good assistant."),886 HumanMessage("what's your name", id="foo", name="example_user"),887 ]888 ```889 """890 messages = convert_to_messages(messages)891 filtered: list[BaseMessage] = []892 for msg in messages:893 if (894 (exclude_names and msg.name in exclude_names)895 or (exclude_types and _is_message_type(msg, exclude_types))896 or (exclude_ids and msg.id in exclude_ids)897 ):898 continue899 900 if exclude_tool_calls is True and (901 (isinstance(msg, AIMessage) and msg.tool_calls)902 or isinstance(msg, ToolMessage)903 ):904 continue905 906 new_msg = msg907 if isinstance(exclude_tool_calls, (list, tuple, set)):908 if isinstance(msg, AIMessage) and msg.tool_calls:909 tool_calls = [910 tool_call911 for tool_call in msg.tool_calls912 if tool_call["id"] not in exclude_tool_calls913 ]914 if not tool_calls:915 continue916 917 content = msg.content918 # handle Anthropic content blocks919 if isinstance(msg.content, list):920 content = [921 content_block922 for content_block in msg.content923 if (924 not isinstance(content_block, dict)925 or content_block.get("type") != "tool_use"926 or content_block.get("id") not in exclude_tool_calls927 )928 ]929 930 new_msg = msg.model_copy(931 update={"tool_calls": tool_calls, "content": content}932 )933 elif (934 isinstance(msg, ToolMessage) and msg.tool_call_id in exclude_tool_calls935 ):936 continue937 938 # default to inclusion when no inclusion criteria given.939 if (940 not (include_types or include_ids or include_names)941 or (include_names and new_msg.name in include_names)942 or (include_types and _is_message_type(new_msg, include_types))943 or (include_ids and new_msg.id in include_ids)944 ):945 filtered.append(new_msg)946 947 return filtered948 949 950@_runnable_support951def merge_message_runs(952 messages: Iterable[MessageLikeRepresentation] | PromptValue,953 *,954 chunk_separator: str = "\n",955) -> list[BaseMessage]:956 r"""Merge consecutive Messages of the same type.957 958 !!! note959 `ToolMessage` objects are not merged, as each has a distinct tool call id that960 can't be merged.961 962 Args:963 messages: Sequence Message-like objects to merge.964 chunk_separator: Specify the string to be inserted between message chunks.965 966 Returns:967 list of BaseMessages with consecutive runs of message types merged into single968 messages. By default, if two messages being merged both have string contents,969 the merged content is a concatenation of the two strings with a new-line970 separator.971 The separator inserted between message chunks can be controlled by specifying972 any string with `chunk_separator`. If at least one of the messages has a list973 of content blocks, the merged content is a list of content blocks.974 975 Example:976 ```python977 from langchain_core.messages import (978 merge_message_runs,979 AIMessage,980 HumanMessage,981 SystemMessage,982 ToolCall,983 )984 985 messages = [986 SystemMessage("you're a good assistant."),987 HumanMessage(988 "what's your favorite color",989 id="foo",990 ),991 HumanMessage(992 "wait your favorite food",993 id="bar",994 ),995 AIMessage(996 "my favorite colo",997 tool_calls=[998 ToolCall(999 name="blah_tool", args={"x": 2}, id="123", type="tool_call"1000 )1001 ],1002 id="baz",1003 ),1004 AIMessage(1005 [{"type": "text", "text": "my favorite dish is lasagna"}],1006 tool_calls=[1007 ToolCall(1008 name="blah_tool",1009 args={"x": -10},1010 id="456",1011 type="tool_call",1012 )1013 ],1014 id="blur",1015 ),1016 ]1017 1018 merge_message_runs(messages)1019 ```1020 1021 ```python1022 [1023 SystemMessage("you're a good assistant."),1024 HumanMessage(1025 "what's your favorite color\\n"1026 "wait your favorite food", id="foo",1027 ),1028 AIMessage(1029 [1030 "my favorite colo",1031 {"type": "text", "text": "my favorite dish is lasagna"}1032 ],1033 tool_calls=[1034 ToolCall({1035 "name": "blah_tool",1036 "args": {"x": 2},1037 "id": "123",1038 "type": "tool_call"1039 }),1040 ToolCall({1041 "name": "blah_tool",1042 "args": {"x": -10},1043 "id": "456",1044 "type": "tool_call"1045 })1046 ]1047 id="baz"1048 ),1049 ]1050 1051 ```1052 """1053 if not messages:1054 return []1055 messages = convert_to_messages(messages)1056 merged: list[BaseMessage] = []1057 for msg in messages:1058 last = merged.pop() if merged else None1059 if not last:1060 merged.append(msg)1061 elif isinstance(msg, ToolMessage) or not isinstance(msg, last.__class__):1062 merged.extend([last, msg])1063 else:1064 last_chunk = _msg_to_chunk(last)1065 curr_chunk = _msg_to_chunk(msg)1066 if curr_chunk.response_metadata:1067 curr_chunk.response_metadata.clear()1068 if (1069 isinstance(last_chunk.content, str)1070 and isinstance(curr_chunk.content, str)1071 and last_chunk.content1072 and curr_chunk.content1073 ):1074 last_chunk.content += chunk_separator1075 merged.append(_chunk_to_msg(last_chunk + curr_chunk))1076 return merged1077 1078 1079# TODO: Update so validation errors (for token_counter, for example) are raised on1080# init not at runtime.1081@_runnable_support1082def trim_messages(1083 messages: Iterable[MessageLikeRepresentation] | PromptValue,1084 *,1085 max_tokens: int,1086 token_counter: Callable[[list[BaseMessage]], int]1087 | Callable[[BaseMessage], int]1088 | BaseLanguageModel1089 | Literal["approximate"],1090 strategy: Literal["first", "last"] = "last",1091 allow_partial: bool = False,1092 end_on: str | type[BaseMessage] | Sequence[str | type[BaseMessage]] | None = None,1093 start_on: str | type[BaseMessage] | Sequence[str | type[BaseMessage]] | None = None,1094 include_system: bool = False,1095 text_splitter: Callable[[str], list[str]] | TextSplitter | None = None,1096) -> list[BaseMessage]:1097 r"""Trim messages to be below a token count.1098 1099 `trim_messages` can be used to reduce the size of a chat history to a specified1100 token or message count.1101 1102 In either case, if passing the trimmed chat history back into a chat model1103 directly, the resulting chat history should usually satisfy the following1104 properties:1105 1106 1. The resulting chat history should be valid. Most chat models expect that chat1107 history starts with either (1) a `HumanMessage` or (2) a `SystemMessage`1108 followed by a `HumanMessage`. To achieve this, set `start_on='human'`.1109 In addition, generally a `ToolMessage` can only appear after an `AIMessage`1110 that involved a tool call.1111 2. It includes recent messages and drops old messages in the chat history.1112 To achieve this set the `strategy='last'`.1113 3. Usually, the new chat history should include the `SystemMessage` if it1114 was present in the original chat history since the `SystemMessage` includes1115 special instructions to the chat model. The `SystemMessage` is almost always1116 the first message in the history if present. To achieve this set the1117 `include_system=True`.1118 1119 !!! note1120 The examples below show how to configure `trim_messages` to achieve a behavior1121 consistent with the above properties.1122 1123 Args:1124 messages: Sequence of Message-like objects to trim.1125 max_tokens: Max token count of trimmed messages.1126 token_counter: Function or llm for counting tokens in a `BaseMessage` or a1127 list of `BaseMessage`.1128 1129 If a `BaseLanguageModel` is passed in then1130 `BaseLanguageModel.get_num_tokens_from_messages()` will be used. Set to1131 `len` to count the number of **messages** in the chat history.1132 1133 You can also use string shortcuts for convenience:1134 1135 - `'approximate'`: Uses `count_tokens_approximately` for fast, approximate1136 token counts.1137 1138 !!! note1139 1140 `count_tokens_approximately` (or the shortcut `'approximate'`) is1141 recommended for using `trim_messages` on the hot path, where exact token1142 counting is not necessary.1143 1144 strategy: Strategy for trimming.1145 1146 - `'first'`: Keep the first `<= n_count` tokens of the messages.1147 - `'last'`: Keep the last `<= n_count` tokens of the messages.1148 allow_partial: Whether to split a message if only part of the message can be1149 included.1150 1151 If `strategy='last'` then the last partial contents of a message are1152 included. If `strategy='first'` then the first partial contents of a1153 message are included.1154 end_on: The message type to end on.1155 1156 If specified then every message after the last occurrence of this type is1157 ignored. If `strategy='last'` then this is done before we attempt to get the1158 last `max_tokens`. If `strategy='first'` then this is done after we get the1159 first `max_tokens`. Can be specified as string names (e.g. `'system'`,1160 `'human'`, `'ai'`, ...) or as `BaseMessage` classes (e.g. `SystemMessage`,1161 `HumanMessage`, `AIMessage`, ...). Can be a single type or a list of types.1162 1163 start_on: The message type to start on.1164 1165 Should only be specified if `strategy='last'`. If specified then every1166 message before the first occurrence of this type is ignored. This is done1167 after we trim the initial messages to the last `max_tokens`. Does not apply1168 to a `SystemMessage` at index 0 if `include_system=True`. Can be specified1169 as string names (e.g. `'system'`, `'human'`, `'ai'`, ...) or as1170 `BaseMessage` classes (e.g. `SystemMessage`, `HumanMessage`, `AIMessage`,1171 ...). Can be a single type or a list of types.1172 1173 include_system: Whether to keep the `SystemMessage` if there is one at index1174 `0`.1175 1176 Should only be specified if `strategy="last"`.1177 text_splitter: Function or `langchain_text_splitters.TextSplitter` for1178 splitting the string contents of a message.1179 1180 Only used if `allow_partial=True`. If `strategy='last'` then the last split1181 tokens from a partial message will be included. if `strategy='first'` then1182 the first split tokens from a partial message will be included. Token1183 splitter assumes that separators are kept, so that split contents can be1184 directly concatenated to recreate the original text. Defaults to splitting1185 on newlines.1186 1187 Returns:1188 List of trimmed `BaseMessage`.1189 1190 Raises:1191 ValueError: if two incompatible arguments are specified or an unrecognized1192 `strategy` is specified.1193 1194 Example:1195 Trim chat history based on token count, keeping the `SystemMessage` if1196 present, and ensuring that the chat history starts with a `HumanMessage` (or a1197 `SystemMessage` followed by a `HumanMessage`).1198 1199 ```python1200 from langchain_core.messages import (