Discord
此笔记本展示了如何创建自己的聊天加载器，该加载器可以将复制粘贴的消息（来自私信）转换为 LangChain 消息列表。
该过程分为四个步骤：
- 通过从 Discord 应用程序复制聊天并将其粘贴到本地计算机上的文件中来创建聊天 .txt 文件
- 从下面复制聊天加载器定义到本地文件中。
- 使用指向文本文件的文件路径初始化
DiscordChatLoader。 - 调用
loader.load()（或loader.lazy_load()）以执行转换。
1. 创建消息转储
目前（2023/08/23），此加载器仅支持以将应用中的消息复制到剪贴板并粘贴到文件中生成的格式的 .txt 文件。以下是一个示例。
%%writefile discord_chats.txt
talkingtower — 08/15/2023 11:10 AM
Love music! Do you like jazz?
reporterbob — 08/15/2023 9:27 PM
Yes! Jazz is fantastic. Ever heard this one?
Website
Listen to classic jazz track...
talkingtower — Yesterday at 5:03 AM
Indeed! Great choice. 🎷
reporterbob — Yesterday at 5:23 AM
Thanks! How about some virtual sightseeing?
Website
Virtual tour of famous landmarks...
talkingtower — Today at 2:38 PM
Sounds fun! Let's explore.
reporterbob — Today at 2:56 PM
Enjoy the tour! See you around.
talkingtower — Today at 3:00 PM
Thank you! Goodbye! 👋
reporterbob — Today at 3:02 PM
Farewell! Happy exploring.
Writing discord_chats.txt
2. 定义聊天加载器
import logging
import re
from typing import Iterator, List
from langchain_community.chat_loaders import base as chat_loaders
from langchain_core.messages import BaseMessage, HumanMessage
logger = logging.getLogger()
class DiscordChatLoader(chat_loaders.BaseChatLoader):
def __init__(self, path: str):
"""
初始化 Discord 聊天加载器。
Args:
path: 导出的 Discord 聊天文本文件的路径。
"""
self.path = path
self._message_line_regex = re.compile(
r"(.+?) — (\w{3,9} \d{1,2}(?:st|nd|rd|th)?(?:, \d{4})? \d{1,2}:\d{2} (?:AM|PM)|Today at \d{1,2}:\d{2} (?:AM|PM)|Yesterday at \d{1,2}:\d{2} (?:AM|PM))",
flags=re.DOTALL,
)
def _load_single_chat_session_from_txt(
self, file_path: str
) -> chat_loaders.ChatSession:
"""
从文本文件加载单个聊天会话。
Args:
file_path: 包含聊天消息的文本文件路径。
Returns:
一个包含加载的聊天消息的 `ChatSession` 对象。
"""
with open(file_path, "r", encoding="utf-8") as file:
lines = file.readlines()
results: List[BaseMessage] = []
current_sender = None
current_timestamp = None
current_content = []
for line in lines:
if re.match(
r".+? — (\d{2}/\d{2}/\d{4} \d{1,2}:\d{2} (?:AM|PM)|Today at \d{1,2}:\d{2} (?:AM|PM)|Yesterday at \d{1,2}:\d{2} (?:AM|PM))",
line,
):
if current_sender and current_content:
results.append(
HumanMessage(
content="".join(current_content).strip(),
additional_kwargs={
"sender": current_sender,
"events": [{"message_time": current_timestamp}],
},
)
)
current_sender, current_timestamp = line.split(" — ")[:2]
current_content = [
line[len(current_sender) + len(current_timestamp) + 4 :].strip()
]
elif re.match(r"\[\d{1,2}:\d{2} (?:AM|PM)\]", line.strip()):
results.append(
HumanMessage(
content="".join(current_content).strip(),
additional_kwargs={
"sender": current_sender,
"events": [{"message_time": current_timestamp}],
},
)
)
current_timestamp = line.strip()[1:-1]
current_content = []
else:
current_content.append("\n" + line.strip())
if current_sender and current_content:
results.append(
HumanMessage(
content="".join(current_content).strip(),
additional_kwargs={
"sender": current_sender,
"events": [{"message_time": current_timestamp}],
},
)
)
return chat_loaders.ChatSession(messages=results)
def lazy_load(self) -> Iterator[chat_loaders.ChatSession]:
"""
懒加载聊天文件中的消息，并以所需格式生成它们。
Yields:
一个包含加载的聊天消息的 `ChatSession` 对象。
"""
yield self._load_single_chat_session_from_txt(self.path)
2. 创建加载器
我们将指向刚刚写入磁盘的文件。
loader = DiscordChatLoader(
path="./discord_chats.txt",
)
3. 加载消息
假设格式正确，加载器将把聊天转换为 langchain 消息。
from typing import List
from langchain_community.chat_loaders.utils import (
map_ai_messages,
merge_chat_runs,
)
from langchain_core.chat_sessions import ChatSession
raw_messages = loader.lazy_load()
# 将来自同一发送者的连续消息合并为一条消息
merged_messages = merge_chat_runs(raw_messages)
# 将“talkingtower”的消息转换为 AI 消息
messages: List[ChatSession] = list(
map_ai_messages(merged_messages, sender="talkingtower")
)
messages
[{'messages': [AIMessage(content='Love music! Do you like jazz?', additional_kwargs={'sender': 'talkingtower', 'events': [{'message_time': '08/15/2023 11:10 AM\n'}]}),
HumanMessage(content='Yes! Jazz is fantastic. Ever heard this one?\nWebsite\nListen to classic jazz track...', additional_kwargs={'sender': 'reporterbob', 'events': [{'message_time': '08/15/2023 9:27 PM\n'}]}),
AIMessage(content='Indeed! Great choice. 🎷', additional_kwargs={'sender': 'talkingtower', 'events': [{'message_time': 'Yesterday at 5:03 AM\n'}]}),
HumanMessage(content='Thanks! How about some virtual sightseeing?\nWebsite\nVirtual tour of famous landmarks...', additional_kwargs={'sender': 'reporterbob', 'events': [{'message_time': 'Yesterday at 5:23 AM\n'}]}),
AIMessage(content="Sounds fun! Let's explore.", additional_kwargs={'sender': 'talkingtower', 'events': [{'message_time': 'Today at 2:38 PM\n'}]}),
HumanMessage(content='Enjoy the tour! See you around.', additional_kwargs={'sender': 'reporterbob', 'events': [{'message_time': 'Today at 2:56 PM\n'}]}),
AIMessage(content='Thank you! Goodbye! 👋', additional_kwargs={'sender': 'talkingtower', 'events': [{'message_time': 'Today at 3:00 PM\n'}]}),
HumanMessage(content='Farewell! Happy exploring.', additional_kwargs={'sender': 'reporterbob', 'events': [{'message_time': 'Today at 3:02 PM\n'}]})]}]
下一步
您可以根据需要使用这些消息，例如微调模型、选择少量示例，或直接为下一个消息进行预测
from langchain_openai import ChatOpenAI
llm = ChatOpenAI()
for chunk in llm.stream(messages[0]["messages"]):
print(chunk.content, end="", flush=True)
Thank you! Have a great day!