MetaGPT + Claude 3 Haiku で、twitterでbookmarkした記事をまとめたい
ということで
先ほど公開した以下の記事は、MetaGPT の examplesのツールを改造して作成したものです。
(原型留めないくらい改造していますが)
毎回前提記事が増えて申し訳ありませんが、以下、2件が前提記事です。
上記で、MetaGPTでHaikuを使える環境を作成した後、以下の手順で実行し、そのまま Zennに貼ればtwitterのまとめメモになるツールを作成しました。
ただ、twitterのbookmarkから情報を吸い出すのは、Chromeの拡張ツールを使っているのですが。
APIで吸い出せないのがツライですね。
これを使っています。お試しで少量のエクスポートができます。
全件エクスポートには、$1を支払う必要がありました。
このツールで、twitterのbookmarkをjson形式ダウンロードします。
そして、pythonで、過去10日間くらいのものを抜き出します。
引数で、何日間を抜き出すか、入力ファイル名、出力ファイル名を指定するようにしてあります。
import json
import argparse
from datetime import datetime, timedelta
def main():
# 引数のパース
parser = argparse.ArgumentParser()
parser.add_argument("--dates", type=int, required=True, help="データ取得日数")
parser.add_argument("--input_path", type=str, required=True, help="入力ファイルパス")
parser.add_argument("--output_path", type=str, required=True, help="出力ファイルパス")
args = parser.parse_args()
# bookmarks.jsonの読み込み
with open(args.input_path, "r", encoding="utf-8") as file:
bookmarks = json.load(file)
# 現在時刻とデータ取得日数前の時刻を計算
current_time = datetime.now()
past_time = current_time - timedelta(days=args.dates)
# 出力配列の初期化
output_data = []
# 各辞書に対する処理
for bookmark in bookmarks:
# tweeted_atを時間の型に変換
tweeted_at = datetime.strptime(bookmark["tweeted_at"], "%Y-%m-%dT%H:%M:%S.%fZ")
# tweeted_atが指定された日数以内であれば出力配列に追加
if tweeted_at > past_time:
# profile_image_url_httpsとextended_mediaを除外して辞書を追加
output_bookmark = {key: value for key, value in bookmark.items() if key not in ["profile_image_url_https", "extended_media"]}
output_data.append(output_bookmark)
# 出力配列をtweeted_atの降順でソート
sorted_output_data = sorted(output_data, key=lambda x: datetime.strptime(x["tweeted_at"], "%Y-%m-%dT%H:%M:%S.%fZ"), reverse=True)
# ソートされた出力配列をJSONとして整形して出力
with open(args.output_path, "w", encoding="utf-8") as file:
json.dump(sorted_output_data, file, indent=4, ensure_ascii=False)
print("データの処理が完了しました。")
if __name__ == "__main__":
main()
実行はこんな感じです。
ryuuri@RTX-3090:~/workspace$ python3 script2.py --input_path /home/ryuuri/workspace/bookmarks.json --output_path /home/ryuuri/workspace/output_data2.json --dates 10
データの処理が完了しました。
ここ10日のものを抜き出し、いらなさそうな、profile_image_url_https, extended_media の情報を削除して、ツイート日付が新しい順に並べ直しています。
この情報を元に、MetaGPTくんのツールで、まとめを作ってもらいます。
HAIKUくんにやって貰っていることは、以下の2点です。
- ツイート本文を40文字程度に翻訳して、タイトルを付けてもらう
- もし、英文だったら、日本語に翻訳してもらう
上のツールで抽出した jsonファイルを入力にして、HAIKUくんの回答を整形して、ファイルに出力しています。
- ※ 2024/3/31 Claude 3 の Rate Limits の更新にあわせて、examples/twitter_bookmarks_to_markdown.py のソースを少し修正しました。
- リクエスト/分(RPM)の制限にも引っかかるので、request数のカウンタを追加しました
- リクエストのカウンタ制限は 5 に設定しました (HAIKU_RPM_LIMIT)
- トークン制限の変数名を HAIKU_TPM_LIMIT に変更しました。
- カウンタを増やすメソッドの中で、カウンタ制限チェックをするようにしました。
- トークン制限を 15,000 に変更しました。
- Rate Limits の変更は以下の通りです。
- https://docs.anthropic.com/claude/reference/rate-limits
- 以下、階層 1 と現在の値を比較しています(現在は階層関係なく同一の値)
- 1分当たりのリクエスト数 50 ⇒ 5
- 1分当たりのトークン数 50,000 ⇒ 25,000
- 1日あたりのトークン数 1,000,000 ⇒ 300,000
- リクエスト/分(RPM)の制限にも引っかかるので、request数のカウンタを追加しました
examples/twitter_bookmarks_to_markdown.py
import json
import argparse
import asyncio
import time
from datetime import datetime, timedelta
from metagpt.logs import logger
from metagpt.llm import LLM
import re
HAIKU_TPM_LIMIT = 15000
HAIKU_RPM_LIMIT = 5
class TokenMeter:
def __init__(self, i_token=0, o_token=0, reqs=0):
self.i_token = i_token
self.o_token = o_token
self.reqs = reqs
self.tpm_limit = HAIKU_TPM_LIMIT
self.rpm_limit = HAIKU_RPM_LIMIT
def clear(self):
self.i_token = 0
self.o_token = 0
self.reqs = 0
def total(self):
return self.i_token + self.o_token
def add(self, msg):
self.i_token = self.i_token + msg.usage.input_tokens
self.o_token = self.o_token + msg.usage.output_tokens
self.reqs = self.reqs + 1
logger.info(f"total i_token: {self.i_token} | total o_token: {self.o_token} | total requests: {self.reqs}")
self.sleep_if_rate_limited()
def sleep_if_rate_limited(self):
if self.total() > self.tpm_limit or self.reqs > self.rpm_limit:
print("トークンの使用制限回避のため、約1分 sleep します...")
self.clear()
time.sleep(65)
class OutputLines:
def __init__(self):
self.lines = []
def clear(self):
self.lines = []
def append(self, line):
self.lines.append(line)
print(line)
def write(self, output_filename):
with open(output_filename, "w", encoding="utf-8") as file:
file.write("\n".join(self.lines))
def is_ascii(string):
pattern = r'^[\x00-\x7F]*$'
return bool(re.match(pattern, string))
async def main():
# 引数のパース
parser = argparse.ArgumentParser()
parser.add_argument("--input_path", type=str, required=True, help="入力ファイルパス")
parser.add_argument("--output_path", type=str, required=True, help="出力ファイルパス")
parser.add_argument("--oembed", action='store_true', help="oembedオプション 自動的に埋め込まれず oembed を使って埋め込むときに指定する")
args = parser.parse_args()
# twitter bookmark jsonの読み込み
with open(args.input_path, "r", encoding="utf-8") as file:
bookmarks = json.load(file)
llm = LLM()
reg = {}
token_meter = TokenMeter()
output_lines = OutputLines()
for dic in bookmarks:
time_stamp = dic.get("tweeted_at", None)
screen_name = dic.get("screen_name", None)
url = dic.get("tweet_url", None)
txt = dic.get("full_text", None)
# twitterのbookmarkツールが、データを取得できていないことがあるので、必要データが全て取れているか確認
if time_stamp and screen_name and url and txt:
if reg.get(url, None):
# 重複排除
continue
reg[url] = time_stamp
if dic.get("note_tweet_text", None):
txt = dic["note_tweet_text"]
msg = [{"role": "user", "content": f"<document>{txt}</document>ドキュメントを簡単に要約して40文字以内のタイトルを付けてください。タイトルだけ回答お 願いします。"}]
result = await llm.acompletion(msg)
token_meter.add(result)
title = result.content[0].text
output_lines.append(f"## {title}")
output_lines.append(f"{time_stamp} {screen_name}")
if args.oembed:
output_lines.append(f"[oembed {url}]")
else:
output_lines.append(f"{url}")
output_lines.append("")
if is_ascii(txt):
msg = [{"role": "user", "content": f"<document>{txt}</document>ドキュメントを日本語に翻訳してください。回答にdocumentタグは含めないでください。"}]
result = await llm.acompletion(msg)
token_meter.add(result)
jp_text = result.content[0].text
output_lines.append(f"{jp_text}")
output_lines.append("")
output_lines.write(args.output_path)
if __name__ == "__main__":
asyncio.run(main())
オプションに、--oembed がありますが、これを付けるとURLを表示するときに、[oembed https://....] という形式で表示します。
knowledgeなどに貼り付けるときに使うことを想定しています。
実行方法は、こんな感じです。
(venv_20240326_MetaGPT) ryuuri@RTX-3090:~/work/MetaGPT/20240326/MetaGPT$ python examples/twitter_bookmarks_to_markdown.py --input_path /home/ryuuri/workspace/output_data2.json --output_path /home/ryuuri/workspace/tweet_summary_zenn.md
これで作成した、/home/ryuuri/workspace/tweet_summary_zenn.md を貼り付けたのが、以下のサンプル記事になります。
公開するかどうかは置いておいて、自分用のメモには便利かなぁと思っています。
--ryuuri/りゅうり/流離
Discussion