Zenn
🔥

MetaGPT + Claude 3 Haiku で、twitterでbookmarkした記事をまとめたい

2024/03/28に公開
Twitter
#
haiku
MetaGPT
#
claude
tech

ということで

先ほど公開した以下の記事は、MetaGPT の examplesのツールを改造して作成したものです。
（原型留めないくらい改造していますが）

https://zenn.dev/ryuuri/articles/54d44d1bfb3052

毎回前提記事が増えて申し訳ありませんが、以下、2件が前提記事です。

https://zenn.dev/ryuuri/articles/4ef61b202f8b9d

https://zenn.dev/ryuuri/articles/e323152e10f5d5

上記で、MetaGPTでHaikuを使える環境を作成した後、以下の手順で実行し、そのまま Zennに貼ればtwitterのまとめメモになるツールを作成しました。

ただ、twitterのbookmarkから情報を吸い出すのは、Chromeの拡張ツールを使っているのですが。
APIで吸い出せないのがツライですね。

https://chromewebstore.google.com/detail/twitterのブックマークエクスポート/fondehlfbfbcegdjhoefhfbkaeengcgd?hl=ja

これを使っています。お試しで少量のエクスポートができます。
全件エクスポートには、$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

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 を貼り付けたのが、以下のサンプル記事になります。

https://zenn.dev/ryuuri/articles/54d44d1bfb3052

公開するかどうかは置いておいて、自分用のメモには便利かなぁと思っています。

--ryuuri／りゅうり／流離

Discussion