There is not yet a straightforward way to export personal WeChat messages. However if you just need no more than few hundreds of messages for model fine-tuning or few-shot examples, this notebook shows how to create your own chat loader that works on copy-pasted WeChat messages to a list of LangChain messages.
Highly inspired by https://python.langchain.com/docs/integrations/chat_loaders/discord
The process has five steps:
- Open your chat in the WeChat desktop app. Select messages you need by mouse-dragging or right-click. Due to restrictions, you can select up to 100 messages once a time.
CMD
/Ctrl
+C
to copy. - Create the chat .txt file by pasting selected messages in a file on your local computer.
- Copy the chat loader definition from below to a local file.
- Initialize the
WeChatChatLoader
with the file path pointed to the text file. - Call
loader.load()
(orloader.lazy_load()
) to perform the conversion.
1. Create message dumpβ
This loader only supports .txt files in the format generated by copying messages in the app to your clipboard and pasting in a file. Below is an example.
%%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
[ε¨η»θ‘¨ζ
]
Overwriting wechat_chats.txt
2. Define chat loaderβ
LangChain currently does not support
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):
"""
Initialize the Discord chat loader.
Args:
path: Path to the exported Discord chat text file.
"""
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))", # noqa
# 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()
# skip non-text messages like stickers, images, etc.
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:
"""
Load a single chat session from a text file.
Args:
file_path: Path to the text file containing the chat messages.
Returns:
A `ChatSession` object containing the loaded chat messages.
"""
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]:
"""
Lazy load the messages from the chat file and yield them in the required format.
Yields:
A `ChatSession` object containing the loaded chat messages.
"""
yield self._load_single_chat_session_from_txt(self.path)
API Reference:
2. Create loaderβ
We will point to the file we just wrote to disk.
loader = WeChatChatLoader(
path="./wechat_chats.txt",
)
3. Load Messagesβ
Assuming the format is correct, the loader will convert the chats to langchain messages.
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()
# Merge consecutive messages from the same sender into a single message
merged_messages = merge_chat_runs(raw_messages)
# Convert messages from "η·ζε" to AI messages
messages: List[ChatSession] = list(map_ai_messages(merged_messages, sender="η·ζε"))
API Reference:
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)]}]
Next Stepsβ
You can then use these messages how you see fit, such as fine-tuning a model, few-shot example selection, or directly make predictions for the next message
from langchain_openai import ChatOpenAI
llm = ChatOpenAI()
for chunk in llm.stream(messages[0]["messages"]):
print(chunk.content, end="", flush=True)