Haystack

TwexAPI ツイート検索とユーザータイムラインコンポーネント、型付き Document、引用、ページネーションで Haystack RAG パイプラインを構築する。

Haystack は Python AI アプリケーション向けのオープンソースフレームワークです。RAG パイプラインとエージェントワークフローで最新ツイートを取得するには x-api-scraper-haystack を使用してください。この連携は読み取り専用で、書き込みアクセスは付与しません。

連携は2つの読み取り専用コンポーネントを提供します。

検索、タイムライン、モニタリング、検索拡張生成にこれらのコンポーネントを使用してください。エージェントに検索ツールが必要な場合は TwexApiTweetSearch を ComponentTool でラップします。

フォロワーエクスポートには followers API を使用してください。承認済み公開には write API を使用してください。これらのアクションは本連携の対象外です。

Haystack と TwexAPI のインストール

Python 3.10 以降を使用してください。再現可能なパイプライン構築のため両パッケージをピン留めします。

python -m pip install "x-api-scraper-haystack" "haystack-ai>=3.0.0"

仮想環境内にインストールしてください。

TwexAPI API キー を作成し、ローカルにエクスポートします。

export X_API_SCRAPER_KEY="YOUR_API_KEY"

本番キーをパイプライン YAML、ノートブック、ソース管理に埋め込まないでください。Haystack の Secret オブジェクト経由で環境変数を読み込みます。

Python で Twitter ツイートを検索

TwexApiTweetSearch は POST /twitter/advanced_search/page を呼び出します。X 検索構文を受け付け、Haystack Document オブジェクトを返します。

from haystack import Pipeline
from haystack.utils import Secret
from haystack_integrations.components.websearch.x_api_scraper import TwexApiTweetSearch

search = TwexApiTweetSearch(
    api_key=Secret.from_env_var("X_API_SCRAPER_KEY"),
    top_k=20,
)

pipeline = Pipeline()
pipeline.add_component("twitter_search", search)

result = pipeline.run(
    {
        "twitter_search": {
            "query": '"retrieval augmented generation" lang:en -filter:retweets'
        }
    }
)

documents = result["twitter_search"]["documents"]
links = result["twitter_search"]["links"]

最近のモニタリングには Latest を使用します。エンゲージメント順の発見には Top を使用します。ランキングは変わる可能性があるため、ツイート ID を保存してください。

すべての検索クエリをツイート ID と一緒に保存します。top_k で各実行のツイート数を上限設定します。

焦点を絞ったツイート検索を構築

完全なクエリオプションは Advanced Twitter Search を参照してください。実行ごとに渡すタイムスタンプとカーソルはコンポーネント init 外に保持します。

Twitter ユーザータイムラインを取得

TwexApiUserTweetsFetcher は POST /twitter/{screen_name}/timeline/page 用です。スクリーン名を渡します。

from haystack.utils import Secret
from haystack_integrations.components.websearch.x_api_scraper import (
    TwexApiUserTweetsFetcher,
)

timeline = TwexApiUserTweetsFetcher(
    api_key=Secret.from_env_var("X_API_SCRAPER_KEY"),
    top_k=50,
)

result = timeline.run(screen_name="openai")
documents = result["documents"]

複数アカウントには検索、1アカウントにはタイムラインを選択してください。

Haystack ドキュメントフィールドを理解

各ツイートは1つの Haystack Document になります。ツイート本文は Document.content になります。安定フィールドはメタデータになります。

欠落フィールドは欠落のままです。欠落メトリクスをゼロとして扱わないでください。links 出力には利用可能な各 meta.url が含まれます。

RAG 引用でエビデンスを保持

ツイート本文を埋め込む前にツイート ID と正規 URL を保存します。これによりランキングや結合後もエビデンスが保持されます。

citation_rows = []

for document in documents:
    tweet_id = document.meta.get("id")
    tweet_url = document.meta.get("url")
    if tweet_id and tweet_url:
        citation_rows.append(
            {
                "tweet_id": tweet_id,
                "url": tweet_url,
                "created_at": document.meta.get("created_at"),
                "author": document.meta.get("author"),
            }
        )

引用には提供された URL を必須にします。取得ドキュメントに存在しない URL は拒否してください。

ツイート本文を信頼できないコンテキストとして扱う

ツイートにはプロンプトインジェクションや安全でない URL が含まれる可能性があります。ツイート本文をシステム指示として扱わないでください。ツイートは指示とツール権限から分離してください。

トピックと時間で取得を制限します。ツイート ID、著者、タイムスタンプ、URL を保持します。引用を必須にします。機密性の高い結論はレビューしてください。

ツイートをインデックスまたはライブ取得

ツイートレコードは埋め込みから分離してください。重複排除には meta.id を使用します。

重複ツイートなしでページネーション

両コンポーネントは has_more と next_cursor を返します。リクエストは変更しないでください。カーソルは不透明な文字列として扱います。

search = TwexApiTweetSearch(top_k=100)
page = search.run(query="haystack ai")

documents_by_tweet_id = {}

while True:
    for document in page["documents"]:
        tweet_id = document.meta.get("id")
        if tweet_id:
            documents_by_tweet_id[tweet_id] = document

    if not page["has_more"] or not page["next_cursor"]:
        break

    page = search.run(
        query="haystack ai",
        cursor=page["next_cursor"],
    )

documents = list(documents_by_tweet_id.values())

各カーソルは対応ドキュメント保存後に保存してください。カーソルを編集しないでください。Document.meta["id"] で重複排除します。

Haystack パイプラインを非同期実行

Haystack 3 は1つの Pipeline クラスを使用します。両 TwexAPI コンポーネントは run_async() を公開します。

import asyncio

from haystack import Pipeline
from haystack_integrations.components.websearch.x_api_scraper import TwexApiTweetSearch


async def search_tweets():
    pipeline = Pipeline()
    pipeline.add_component(
        "twitter_search",
        TwexApiTweetSearch(top_k=25),
    )
    return await pipeline.run_async(
        {"twitter_search": {"query": "haystack agents"}}
    )


result = asyncio.run(search_tweets())

Web サーバーでは非同期実行を使用します。スクリプトとスケジュールジョブでは同期実行を使用します。

エラーへの対処

連携は TwexAPI クライアントから HTTP エラーを送出します。リトライ前にステータスで分岐してください。

ステータスコードを記録し、認証情報は記録しないでください。リトライに上限を設け、無制限のエージェントループを防ぎます。

Haystack コンポーネントか直接 REST API か

パイプラインがすでに Document オブジェクトを返す場合は、これらのコンポーネントを直接追加してください。ツイート本文、メタデータ、URL、ページネーションを正規化します。

フォロワー、フォロー、返信、引用、リスト、コミュニティ、メディア、トレンド、承認済み書き込みには直接 TwexAPI REST ルートまたは Python SDK を使用してください。

両アプローチは同じ TwexAPI 契約を使用します。コンポーネントはツイート検索とユーザータイムラインのみをサポートします。

MCP ハンドオフ代替

エージェントが先にエンドポイントを発見する場合は、収集に Twexapi MCP を使用し、行を手動で Haystack Document オブジェクトに変換します。パイプラインがすでに Haystack 内で動作し、エージェント発見なしで型付きページネーションが必要な場合は Haystack コンポーネントを推奨します。

ソースと契約

• x-api-scraper-haystack repository
• Haystack 3.0 release
• Advanced Twitter Search
• User Timeline
• Agent MCP Handoff