はじめに

LangChainを用いて、trendspyからトレンドワードを取得し、トレンドへの上昇理由をTwitter APIで取得したツイートから考察するエージェントを作成しました。

用意したToolは以下の2つです
・trendspyを用いてトレンドワードを取得する関数
・Twitter APIを用いて検索ワードが含まれる最新のツイートを10件取得する関数

環境準備

パッケージは以下のものを利用します。

langchain==0.3.15
langchain-community==0.3.15
langchain-openai==0.3.1
trendspy==0.1.6
tweepy==4.15.0

Xの開発者アカウントを作成します。今回は無料プランで行いました。APIのアクセス制限が15分に10回と記載されていましたが、登録した直後だと15分に1回までだったので注意が必要です。また取得できるツイート数も1ヶ月に100件までです。

「Bearer Token」の項目からトークンを取得します。

取得したキーとopenAIのキーは.envに記述します。

.env

OPENAI_API_KEY="OPENAI_API_KEY"
BEARER_TOKEN="BEARER_TOKEN"

以下がコードの全体像です。

from dotenv import load_dotenv
from tweet_search_api import get_tweets  # 外部API呼び出し関数をインポート
from trendspy_api import get_japan_trends
from langchain_openai import ChatOpenAI  # ChatOpenAIを使用
from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType

# .envファイルを読み込む
load_dotenv()

# 外部APIツールを定義
tools = [
    Tool(
        name        = "Tweet_Search_API",
        func        = get_tweets,
        description = "このツールは検索ワードを引数として渡すことで、JSONで10件のツイートを取得することができます。"
    ),
    Tool(
        name        = "Trend_Get_API",
        func        = get_japan_trends,
        description = "この関数は、現在の日本のトレンドワード1つをJSONで取得することができます。"
    )
]

# ChatOpenAIの設定（APIキーを環境変数から取得）
openai_model_version = "gpt-4o-mini"
llm                  = ChatOpenAI(model=openai_model_version)

# エージェントの初期化
agent_executor = initialize_agent(
    tools, llm, agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True
)

# クエリを辞書形式で渡す
query = {"messages": [{"role": "user", "content": "現在の日本のトレンドワードを教えて。そのトレンドワードがなぜトレンドに入っているのか最新のツイートをもとに教えて"}]}

# エージェントにクエリを送信
result = agent_executor.invoke(query["messages"][0]["content"])

# 結果を表示
print(result)

また、trendspyを用いたトレンドワードの取得は以下になります。

import json
from trendspy import Trends

def get_japan_trends(props):
    # Trendsオブジェクトの作成
    tr = Trends()

    # 日本の現在のトレンド検索を取得
    trends = tr.trending_now(geo='JP')

    # トレンドデータをJSON形式に変換
    trend_list = [
        {"keyword": trends[0].keyword}  # 最新のトレンドワードを取得
    ]

    # JSON出力
    return json.dumps(trend_list, ensure_ascii=False, indent=4)

Tweetの取得を行う関数は以下の通りです。

import json
from dotenv import load_dotenv
import os
import tweepy

# 環境変数からAPIキーを取得
load_dotenv()

BEARER_TOKEN = os.getenv("BEARER_TOKEN")

# Tweepy クライアントを作成
client = tweepy.Client(bearer_token=BEARER_TOKEN)

def get_tweets(query):
    """検索ワードのみを受け取り、ツイートを10件取得する"""
    try:
        # APIリクエスト（リツイートを除外、最大10件）
        tweets = client.search_recent_tweets(
            query=f"{query} -is:retweet -has:links",
            max_results=10,
            tweet_fields=["created_at", "author_id"]
        )

        result = {
            "query": query,
            "count": 10,
            "tweets": []
        }

        if tweets.data:
            for tweet in tweets.data:
                result["tweets"].append({
                    "id": tweet.id,
                    "text": tweet.text,
                    "author_id": tweet.author_id,
                    "created_at": tweet.created_at.isoformat()
                })

        return json.dumps(result, ensure_ascii=False)

    except Exception as e:
        return json.dumps({"error": str(e)}, ensure_ascii=False)

以下のような回答が得られます。

{'input': '現在の日本のトレンドワードを教えて。そのトレンドワードがなぜトレンドに入っているのか最新のツイートをもとに教えて', 'output': '現在の日本のトレンドワードは「大船渡 山火事」です。これは岩手県の大船渡市で起きた大規模な山火事に関連しており、80件近い家屋が焼失していることから、多くの人々が心配している状況です。'}

最後に

trendspyによるトレンドワードの取得の後に、Twitter APIによるツイートの取得を行いトレンドへの掲載理由を回答することができました。2つのツールの使用順番などを指定せずとも使いこなし理由を考察してくれるのが魅力的だと思いました。