終極需求：做一個專屬於我的知識庫。這個嘛，只是這個終極需求的第一步。這個僅僅是作為 demo，後面還有如山一樣多的事情要做。
(先來一個最終的效果圖看看)
第一步 - 導出 Telegram 頻道資訊#
在開始之前，你需要確保已經安裝 Python 3。你還需要以下內容：
- Telethon、PySocks 庫：可以使用
pip install telethon PySocks安裝。 - 確保你是你想要獲取消息的頻道的成員。
- 一個有效的 Telegram 帳號，獲取 Telegram 應用程序的 API ID 和 API hash，你可以在 https://my.telegram.org 獲取。(保護好你的 API 密鑰，不要在公共倉庫或公開場合洩露。)
- 代理伺服器（可選，如果你處於 gfw 之內的話）
代碼實現 - 導出頻道資訊#
- 將下面的 Python 腳本保存為 telegram_to_csv.py：
import csv
import socks
from telethon import TelegramClient
from telethon.tl.functions.messages import GetHistoryRequest
# 設置 TelegramClient，連接到 Telegram API
client = TelegramClient(
'demo',
'api_id',
'api_hash',
proxy=(socks.SOCKS5, '127.0.0.1', 1080)
)
async def export_to_csv(filename, fieldnames, data):
"""
將數據導出到 CSV 文件中。
參數:
filename -- 導出文件的名稱
fieldnames -- CSV 頭部字段名稱列表
data -- 要導出的字典列表
"""
with open(filename, 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(data)
async def fetch_messages(channel_username):
"""
獲取指定頻道的所有消息。
參數:
channel_username -- 目標頻道的用戶名
"""
channel_entity = await client.get_input_entity(channel_username)
offset_id = 0 # 初始消息 ID 偏移量
all_messages = [] # 存儲所有消息的列表
while True:
# 請求消息記錄
history = await client(GetHistoryRequest(
peer=channel_entity,
offset_id=offset_id,
offset_date=None,
add_offset=0,
limit=100, # 每次請求的消息數量
max_id=0,
min_id=0,
hash=0
))
if not history.messages: # 當沒有更多消息時結束循環
break
for message in history.messages:
if message.message: # 僅處理有文本內容的消息
# 將消息序列化為字典形式
message_dict = {
'id': message.id,
'date': message.date.strftime('%Y-%m-%d %H:%M:%S'),
'text': message.message
}
all_messages.append(message_dict)
offset_id = history.messages[-1].id
print(f"Fetched messages: {len(all_messages)}")
return all_messages
async def main():
"""
主程序：從指定頻道獲取消息並保存到 CSV 文件中。
"""
await client.start() # 啟動 Telegram 客戶端
print("Client Created")
channel_username = 'niracler_channel' # 你要抓取的 Telegram 頻道用戶名
all_messages = await fetch_messages(channel_username) # 獲取消息
# 定義 CSV 文件的頭部，並導出
headers = ['id', 'date', 'text']
await export_to_csv('channel_messages.csv', headers, all_messages)
# 當該腳本作為主程序運行時
if __name__ == '__main__':
client.loop.run_until_complete(main())
運行腳本 telegram_to_csv.py#
在終端運行腳本：
python telegram_to_csv.py
腳本會開始運行，並將來自指定 Telegram 頻道的所有消息保存到當前目錄下名為 channel_messages.csv 的文件中。
完成以上步驟後，你將在 channel_messages.csv 文件中找到頻道內的文本消息，包括消息的 ID、日期及其內容。
(結果就不貼出來了～～)
第二步 - 使用 openai 的 text-embedding-ada-002 模型對文本進行 embedding#
- 安裝 openai、pandas 庫，可以使用
pip install openai pandas安裝。 - 一個有效的 OpenAI API 密鑰
代碼實現 - embedding#
將下面的 Python 腳本保存為 embedding_generator.py：
import pandas as pd
from openai import OpenAI
# 配置 OpenAI 客戶端
client = OpenAI(api_key='YOUR_API_KEY')
def get_embedding(text, model="text-embedding-ada-002"):
"""
獲取文本的嵌入向量。
"""
text = text.replace("\n", " ") # 清理文本中的換行符
response = client.embeddings.create(input=[text], model=model) # 請求嵌入向量
return response.data[0].embedding # 提取和返回嵌入向量
def embedding_gen():
"""
生成教程文本嵌入向量數據。
"""
df = pd.read_csv('channel_messages.csv') # 讀取 CSV 文件到 DataFrame
df['text_with_date'] = df['date'] + " " + df['text'] # 拼接日期和文本
df['ada_embedding'] = df[:100].text_with_date.apply(get_embedding) # 批量應用文本嵌入函數
del df['text_with_date'] # 刪除 'text_with_date' 列
df.to_csv('embedded_1k_reviews.csv', index=False) # 保存結果到新的 CSV 文件
# 打印 DataFrame 的前幾行進行確認
print(df.head())
# 當腳本被直接運行時
if __name__ == "__main__":
embedding_gen()
運行腳本#
python embedding_generator.py
第三步 - 進行搜索#
- 安裝 pandas、numpy、tabulate 庫，可以使用
pip install pandas numpy tabulate安裝。 - tabulate 庫用於將 DataFrame 打印為表格形式。
代碼實現 - 搜索#
將下面的 Python 腳本保存為 embedding_search.py：
import ast
import sys
import pandas as pd
import numpy as np
from tabulate import tabulate
from openai import OpenAI
# 配置 OpenAI 客戶端
client = OpenAI(api_key='YOUR_API_KEY')
def get_embedding(text, model="text-embedding-ada-002"):
"""
獲取文本的嵌入向量。
"""
text = text.replace("\n", " ") # 清理文本中的換行符
response = client.embeddings.create(input=[text], model=model) # 請求嵌入向量
return response.data[0].embedding # 提取和返回嵌入向量
def cosine_similarity(a, b):
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
def embedding_search(query, df, model="text-embedding-ada-002"):
"""
使用 OpenAI API 搜索嵌入向量。
"""
query_embedding = get_embedding(query, model=model) # 獲取查詢文本的嵌入向量
df['similarity'] = df.ada_embedding.apply(lambda x: cosine_similarity(ast.literal_eval(x), query_embedding)) # 計算相似度
df = df.sort_values(by='similarity', ascending=False) # 按相似度降序排列
df = df.drop(columns=['ada_embedding']) # 刪除嵌入向量列
return df
if __name__ == "__main__":
df = pd.read_csv('embedded_1k_reviews.csv') # 讀取 CSV 文件到 DataFrame
query = sys.argv[1]
df = embedding_search(query, df) # 搜索嵌入向量
print(tabulate(df.head(10), headers='keys', tablefmt='psql')) # 打印前 10 條結果
運行腳本 - 結果#
$ python embedding_search.py 動物森友會
+------+---------------------+--------------------------------------------------------------+----------------+
| | date | text | similarities |
|------+---------------------+--------------------------------------------------------------+----------------|
| 1041 | 2021-04-03 06:18:40 | 尼爾 森友會 | 0.843896 |
| 836 | 2021-10-16 02:37:16 | 動森直面會中文視頻 | 0.826405 |
| | | https://www.youtube.com/watch?v=rI_jWfNd2dc | |
| 1208 | 2019-11-10 00:05:56 | 雲養動物好像很有趣啊 | 0.822377 |
| 489 | 2023-06-16 09:33:15 | 看著小貓的生活就想起西西佛神話裡面的西西佛 | 0.802677 |
| 369 | 2023-08-16 02:15:54 | 家貓會不會無聊寂寞 | 0.797062 |
| 13 | 2023-12-14 13:17:59 | 參加了🤗 | 0.796492 |
| 1177 | 2020-02-12 10:27:45 | 國人吃野生動物這種事情之所以屢屢不絕，和頭腦中根深蒂固的中醫 | 0.796363 |
| | | 觀念亦有直接關係。養生、食療、進補、藥膳、以形補形、補氣血…… | |
| | | 偽科學死灰復燃，現在不加以遏制，以後也還會發生類似的事情。 | |
| | | 科學才是唯一的道路。 | |
| 801 | 2021-11-07 13:46:21 | 想不到，今年的年度遊戲竟然還是動森跟火紋。 | 0.796246 |
| | | 動森是之前沒玩夠，火紋是因為某個最大最惡事件導致我要重玩的。 | |
| 837 | 2021-10-16 02:37:16 | 不會吧，難道我的年度遊戲又要變成動森了嗎？ | 0.795871 |
| 423 | 2023-07-29 14:11:22 | 鬼畜到能稱之為精神污染的頭像了～～ | 0.794144 |
+------+---------------------+--------------------------------------------------------------+----------------+
路漫漫其修遠兮 - 後面要做的事情還多著呢#
- 向量數據庫: 機器人就可以使用這個向量數據庫來進行搜索，每次用 csv 文件太低效了。有在考慮用 cloudflare 的 vectorize。只不過想着要先做個簡單的實驗了解流程再說。畢竟 cloudflare 的 paid plan 才能用 vectorize，而且我也不知道這個功能是否能夠滿足我的需求。
- 後續持續更新數據庫: 不僅僅是我的頻道，還有我的文章什麼的也是相應的數據源，甚至是一些我關注的頻道，而且用 telegram bot 機器人持續自動更新數據庫。
- Prompt Engineering: 向 chatgpt 提問時，可以從這個向量數據庫中找出相關內容，然後一起拼到 prompt 去問 chatgpt。
- 基礎知識: 我不能等到基礎知識夠了再去幹這些事情吧，我應該是要邊幹邊學的。現在已經將我理解的幾步給做了，後面要補充一下對應的知識儲備。
- 提高質量: 一些低質量的內容不應該放進去，而且要儘量減少一些圖片相關的內容才行，因為圖片是不能被 embedding 的。
- 做成 CLI: 其實這個功能是寫在 奈亞子 CLI 裡面，但是代碼還沒整理好，也完全沒有異常處理，所以先作為 demo 放出來。在這裡貼這麼長串的代碼效果也是不太好。
參考資料#
- Embedding paragraphs from my blog with E5-large-v2 - 我做這個事情的初心就是這篇文章，不過基本也就看個思路，畢竟我 embedding 是直接用的 openai 的 api，而不是本地模型
- telethon 的文檔 - 這是一個個人帳號的 telegram api 的 python 封裝，因為個人帳號用的是 mtproto 協議，所以不能簡單地調用接口，使用這個庫的必要程度還是很高的
- openai embeddings use cases - 我就是照著這個例子做 embedding 的
後記#
也是一個沒有什麼技術含量的文章，算是記錄了我又學到了一些東西吧。