微信
目前还没有简单的方法来导出个人微信消息。不过,如果您只需要几百条消息用于模型微调或少量示例,这个笔记本展示了如何创建自己的聊天加载器,该加载器可以处理复制并粘贴的微信消息,将其转换为 LangChain 消息列表。
深受 https://python.langchain.com/docs/integrations/chat_loaders/discord 启发
这个过程分为五个步骤:
- 在微信桌面应用中打开您的聊天。通过鼠标拖动或右键选择所需的消息。由于限制,您一次最多可以选择 100 条消息。
CMD
/Ctrl
+C
进行复制。 - 通过将选定的消息粘贴到您本地计算机上的文件中来创建聊天 .txt 文件。
- 从下面复制聊天加载器的定义到本地文件中。
- 使用指向文本文件的文件路径初始化
WeChatChatLoader
。 - 调用
loader.load()
(或loader.lazy_load()
)来执行转换。
1. 创建消息转储
该加载器仅支持以 .txt 文件格式生成的文件,该格式通过将应用中的消息复制到剪贴板并粘贴到文件中。下面是一个示例。
%%writefile wechat_chats.txt
女朋友 2023/09/16 2:51 PM
天气有点凉
男朋友 2023/09/16 2:51 PM
珍簟凉风著,瑶琴寄恨生。嵇君懒书札,底物慰秋情。
女朋友 2023/09/16 3:06 PM
忙什么呢
男朋友 2023/09/16 3:06 PM
今天只干成了一件像样的事
那就是想你
女朋友 2023/09/16 3:06 PM
[动画表情]
正在覆盖 wechat_chats.txt
2. 定义聊天加载器
LangChain 当前不支持
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 WeChatChatLoader(chat_loaders.BaseChatLoader):
def __init__(self, path: str):
"""
初始化 Discord 聊天加载器。
Args:
path: 导出的 Discord 聊天文本文件的路径。
"""
self.path = path
self._message_line_regex = re.compile(
r"(?P<sender>.+?) (?P<timestamp>\d{4}/\d{2}/\d{2} \d{1,2}:\d{2} (?:AM|PM))",
# flags=re.DOTALL,
)
def _append_message_to_results(
self,
results: List,
current_sender: str,
current_timestamp: str,
current_content: List[str],
):
content = "\n".join(current_content).strip()
# 跳过非文本消息,如贴纸、图片等。
if not re.match(r"\[.*\]", content):
results.append(
HumanMessage(
content=content,
additional_kwargs={
"sender": current_sender,
"events": [{"message_time": current_timestamp}],
},
)
)
return results
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(self._message_line_regex, line):
if current_sender and current_content:
results = self._append_message_to_results(
results, current_sender, current_timestamp, current_content
)
current_sender, current_timestamp = re.match(
self._message_line_regex, line
).groups()
current_content = []
else:
current_content.append(line.strip())
if current_sender and current_content:
results = self._append_message_to_results(
results, current_sender, current_timestamp, current_content
)
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 = WeChatChatLoader(
path="./wechat_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)
# 将消息从 "男朋友" 转换为 AI 消息
messages: List[ChatSession] = list(map_ai_messages(merged_messages, sender="男朋友"))
messages
[{'messages': [HumanMessage(content='天气有点凉', additional_kwargs={'sender': '女朋友', 'events': [{'message_time': '2023/09/16 2:51 PM'}]}, example=False),
AIMessage(content='珍簟凉风著,瑶琴寄恨生。嵇君懒书札,底物慰秋情。', additional_kwargs={'sender': '男朋友', 'events': [{'message_time': '2023/09/16 2:51 PM'}]}, example=False),
HumanMessage(content='忙什么呢', additional_kwargs={'sender': '女朋友', 'events': [{'message_time': '2023/09/16 3:06 PM'}]}, example=False),
AIMessage(content='今天只干成了一件像样的事\n那就是想你', additional_kwargs={'sender': '男朋友', 'events': [{'message_time': '2023/09/16 3:06 PM'}]}, example=False)]}]
下一步
您可以根据需要使用这些消息,例如微调模型、选择少量示例,或直接进行下一条消息的预测。
from langchain_openai import ChatOpenAI
llm = ChatOpenAI()
for chunk in llm.stream(messages[0]["messages"]):
print(chunk.content, end="", flush=True)