Source code for langchain_community.chat_loaders.utils
"""用于聊天加载程序的工具。"""
from copy import deepcopy
from typing import Iterable, Iterator, List
from langchain_core.chat_sessions import ChatSession
from langchain_core.messages import AIMessage, BaseMessage
[docs]def merge_chat_runs_in_session(
chat_session: ChatSession, delimiter: str = "\n\n"
) -> ChatSession:
"""将聊天会话中的聊天运行合并在一起。
聊天运行是来自同一发送者的消息序列。
参数:
chat_session:一个聊天会话。
返回:
一个合并了聊天运行的聊天会话。
"""
messages: List[BaseMessage] = []
for message in chat_session["messages"]:
if not isinstance(message.content, str):
raise ValueError(
"Chat Loaders only support messages with content type string, "
f"got {message.content}"
)
if not messages:
messages.append(deepcopy(message))
elif (
isinstance(message, type(messages[-1]))
and messages[-1].additional_kwargs.get("sender") is not None
and messages[-1].additional_kwargs["sender"]
== message.additional_kwargs.get("sender")
):
if not isinstance(messages[-1].content, str):
raise ValueError(
"Chat Loaders only support messages with content type string, "
f"got {messages[-1].content}"
)
messages[-1].content = (
messages[-1].content + delimiter + message.content
).strip()
messages[-1].additional_kwargs.get("events", []).extend(
message.additional_kwargs.get("events") or []
)
else:
messages.append(deepcopy(message))
return ChatSession(messages=messages)
[docs]def merge_chat_runs(chat_sessions: Iterable[ChatSession]) -> Iterator[ChatSession]:
"""合并聊天记录。
聊天记录是来自同一发送者的消息序列。
参数:
chat_sessions:聊天会话的列表。
返回:
合并后的聊天会话列表。
"""
for chat_session in chat_sessions:
yield merge_chat_runs_in_session(chat_session)
[docs]def map_ai_messages_in_session(chat_sessions: ChatSession, sender: str) -> ChatSession:
"""将来自指定“发送者”的消息转换为AI消息。
这对于微调AI以适应您的语音非常有用。
"""
messages = []
num_converted = 0
for message in chat_sessions["messages"]:
if message.additional_kwargs.get("sender") == sender:
message = AIMessage(
content=message.content,
additional_kwargs=message.additional_kwargs.copy(),
example=getattr(message, "example", None), # type: ignore[arg-type]
)
num_converted += 1
messages.append(message)
return ChatSession(messages=messages)
[docs]def map_ai_messages(
chat_sessions: Iterable[ChatSession], sender: str
) -> Iterator[ChatSession]:
"""将来自指定“发送者”的消息转换为AI消息。
这对于微调AI以适应您的语音非常有用。
"""
for chat_session in chat_sessions:
yield map_ai_messages_in_session(chat_session, sender)