from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
from langchain_core.messages.ai import (
AIMessage,
AIMessageChunk,
)
from langchain_core.messages.base import (
BaseMessage,
BaseMessageChunk,
)
from langchain_core.messages.chat import ChatMessage, ChatMessageChunk
from langchain_core.messages.function import FunctionMessage, FunctionMessageChunk
from langchain_core.messages.human import HumanMessage, HumanMessageChunk
from langchain_core.messages.system import SystemMessage, SystemMessageChunk
from langchain_core.messages.tool import ToolMessage, ToolMessageChunk
AnyMessage = Union[
AIMessage, HumanMessage, ChatMessage, SystemMessage, FunctionMessage, ToolMessage
]
[docs]def get_buffer_string(
messages: Sequence[BaseMessage], human_prefix: str = "Human", ai_prefix: str = "AI"
) -> str:
"""将一系列消息转换为字符串并将它们连接成一个字符串。
Args:
messages: 要转换为字符串的消息。
human_prefix: 要添加到HumanMessages内容前面的前缀。
ai_prefix: 要添加到AIMessages内容前面的前缀。
Returns:
所有输入消息的单个字符串连接。
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?
AI: Good, how are you?"
"""
string_messages = []
for m in messages:
if isinstance(m, HumanMessage):
role = human_prefix
elif isinstance(m, AIMessage):
role = ai_prefix
elif isinstance(m, SystemMessage):
role = "System"
elif isinstance(m, FunctionMessage):
role = "Function"
elif isinstance(m, ToolMessage):
role = "Tool"
elif isinstance(m, ChatMessage):
role = m.role
else:
raise ValueError(f"Got unsupported message type: {m}")
message = f"{role}: {m.content}"
if isinstance(m, AIMessage) and "function_call" in m.additional_kwargs:
message += f"{m.additional_kwargs['function_call']}"
string_messages.append(message)
return "\n".join(string_messages)
def _message_from_dict(message: dict) -> BaseMessage:
_type = message["type"]
if _type == "human":
return HumanMessage(**message["data"])
elif _type == "ai":
return AIMessage(**message["data"])
elif _type == "system":
return SystemMessage(**message["data"])
elif _type == "chat":
return ChatMessage(**message["data"])
elif _type == "function":
return FunctionMessage(**message["data"])
elif _type == "tool":
return ToolMessage(**message["data"])
elif _type == "AIMessageChunk":
return AIMessageChunk(**message["data"])
elif _type == "HumanMessageChunk":
return HumanMessageChunk(**message["data"])
elif _type == "FunctionMessageChunk":
return FunctionMessageChunk(**message["data"])
elif _type == "ToolMessageChunk":
return ToolMessageChunk(**message["data"])
elif _type == "SystemMessageChunk":
return SystemMessageChunk(**message["data"])
elif _type == "ChatMessageChunk":
return ChatMessageChunk(**message["data"])
else:
raise ValueError(f"Got unexpected message type: {_type}")
[docs]def messages_from_dict(messages: Sequence[dict]) -> List[BaseMessage]:
"""将字典格式的消息序列转换为消息对象。
参数：
messages：需要转换的消息序列（以字典形式）。
返回：
消息列表（BaseMessages）。
"""
return [_message_from_dict(m) for m in messages]
[docs]def message_chunk_to_message(chunk: BaseMessageChunk) -> BaseMessage:
"""将消息块转换为消息。
参数：
chunk：要转换的消息块。
返回：
消息。
"""
if not isinstance(chunk, BaseMessageChunk):
return chunk
# chunk classes always have the equivalent non-chunk class as their first parent
ignore_keys = ["type"]
if isinstance(chunk, AIMessageChunk):
ignore_keys.append("tool_call_chunks")
return chunk.__class__.__mro__[1](
**{k: v for k, v in chunk.__dict__.items() if k not in ignore_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:
"""根据消息类型和内容字符串创建一条消息。
参数：
message_type: str 消息类型（例如，"human"，"ai"等）。
content: str 内容字符串。
返回：
适当类型的消息。
"""
kwargs: Dict[str, Any] = {}
if name is not None:
kwargs["name"] = name
if tool_call_id is not None:
kwargs["tool_call_id"] = tool_call_id
if additional_kwargs:
kwargs["additional_kwargs"] = additional_kwargs # type: ignore[assignment]
if id is not None:
kwargs["id"] = id
if tool_calls is not None:
kwargs["tool_calls"] = tool_calls
if message_type in ("human", "user"):
message: BaseMessage = HumanMessage(content=content, **kwargs)
elif message_type in ("ai", "assistant"):
message = AIMessage(content=content, **kwargs)
elif message_type == "system":
message = SystemMessage(content=content, **kwargs)
elif message_type == "function":
message = FunctionMessage(content=content, **kwargs)
elif message_type == "tool":
message = ToolMessage(content=content, **kwargs)
else:
raise ValueError(
f"Unexpected message type: {message_type}. Use one of 'human',"
f" 'user', 'ai', 'assistant', or 'system'."
)
return message
def _convert_to_message(
message: MessageLikeRepresentation,
) -> BaseMessage:
"""实例化各种消息格式的消息。
消息格式可以是以下之一：
- BaseMessagePromptTemplate
- BaseMessage
- 2元组（角色字符串，模板）；例如，("human", "{user_input}")
- 字典：带有角色和内容键的消息字典
- 字符串：简写形式为("human", template)；例如，"{user_input}"
参数：
message：支持格式之一的消息表示
返回：
消息实例或消息模板的实例
"""
if isinstance(message, BaseMessage):
_message = message
elif isinstance(message, str):
_message = _create_message_from_message_type("human", message)
elif isinstance(message, Sequence) and len(message) == 2:
# mypy doesn't realise this can't be a string given the previous branch
message_type_str, template = message # type: ignore[misc]
_message = _create_message_from_message_type(message_type_str, template)
elif isinstance(message, dict):
msg_kwargs = message.copy()
try:
try:
msg_type = msg_kwargs.pop("role")
except KeyError:
msg_type = msg_kwargs.pop("type")
msg_content = msg_kwargs.pop("content")
except KeyError:
raise ValueError(
f"Message dict must contain 'role' and 'content' keys, got {message}"
)
_message = _create_message_from_message_type(
msg_type, msg_content, **msg_kwargs
)
else:
raise NotImplementedError(f"Unsupported message type: {type(message)}")
return _message
[docs]def convert_to_messages(
messages: Sequence[MessageLikeRepresentation],
) -> List[BaseMessage]:
"""将一系列消息转换为消息列表。
参数：
messages：要转换的消息序列。
返回：
消息列表（BaseMessages）。
"""
return [_convert_to_message(m) for m in messages]