Haystack

Construisez des pipelines RAG Haystack avec les composants de recherche de tweets et de timelines TwexAPI, des Documents typés, des citations et la pagination.

Haystack est un framework open source pour les applications IA en Python. Utilisez x-api-scraper-haystack pour des tweets récents dans des pipelines RAG et des workflows d’agents. L’intégration est en lecture seule et ne donne jamais accès à l’écriture.

L’intégration fournit deux composants en lecture seule :

Utilisez ces composants pour la recherche, les timelines, la surveillance et la génération augmentée par la récupération. Enveloppez TwexApiTweetSearch avec ComponentTool lorsqu’un agent a besoin d’un outil de recherche.

Utilisez l’API followers pour les exports d’abonnés. Utilisez l’API d’écriture pour la publication approuvée. Ces actions restent en dehors de cette intégration.

Installer Haystack et TwexAPI

Utilisez Python 3.10 ou plus récent. Verrouillez les deux packages pour des builds de pipeline reproductibles.

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

Installez dans un environnement virtuel.

Créez une clé API TwexAPI, puis exportez-la localement.

export X_API_SCRAPER_KEY="YOUR_API_KEY"

Ne jamais intégrer des clés de production dans le YAML de pipeline, des notebooks ou le contrôle de version. Chargez la variable d’environnement via un objet Haystack Secret.

Rechercher des tweets Twitter en Python

TwexApiTweetSearch appelle POST /twitter/advanced_search/page. Il accepte la syntaxe de recherche X et renvoie des objets 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"]

Utilisez Latest pour la surveillance récente. Utilisez Top pour la découverte classée par engagement. Stockez les ID de tweet car les classements peuvent changer.

Enregistrez chaque requête de recherche à côté de ses ID de tweet. Définissez top_k pour limiter le nombre de tweets à chaque exécution.

Construire des recherches de tweets ciblées

Consultez Advanced Twitter Search pour toutes les options de requête. Gardez les horodatages et curseurs en dehors de l’init du composant lorsque vous les passez par exécution.

Récupérer une timeline utilisateur Twitter

Utilisez TwexApiUserTweetsFetcher pour POST /twitter/{screen_name}/timeline/page. Passez un nom d’écran.

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"]

Choisissez la recherche pour plusieurs comptes et la timeline pour un seul compte.

Comprendre les champs de document Haystack

Chaque tweet devient un Document Haystack. Le texte du tweet devient Document.content. Les champs stables deviennent des métadonnées.

Les champs manquants restent absents. Ne jamais traiter des métriques manquantes comme zéro. La sortie links contient chaque meta.url disponible.

Conserver les preuves dans les citations RAG

Stockez les ID de tweet et les URLs canoniques avant d’embarquer le texte du tweet. Cela préserve les preuves après classement ou jointure.

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"),
            }
        )

Exigez des URLs fournies pour les citations. Rejetez les URLs absentes des documents récupérés.

Traiter le texte de tweet comme contexte non fiable

Les tweets peuvent contenir des injections de prompt et des URLs non sûres. Ne jamais traiter le texte de tweet comme une instruction système. Gardez les tweets séparés des instructions et des permissions d’outils.

Limitez la récupération par sujet et période. Préservez les ID de tweet, auteurs, horodatages et URLs. Exigez des citations. Revoyez les conclusions sensibles.

Indexer les tweets ou les récupérer en direct

Gardez les enregistrements de tweet séparés des embeddings. Utilisez meta.id pour la déduplication.

Paginer sans tweets en double

Les deux composants renvoient has_more et next_cursor. Gardez la requête inchangée. Traitez les curseurs comme des chaînes opaques.

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())

Enregistrez chaque curseur après ses documents. Ne jamais modifier un curseur. Dédupliquez sur Document.meta["id"].

Exécuter des pipelines Haystack en asynchrone

Haystack 3 utilise une seule classe Pipeline. Les deux composants TwexAPI exposent 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())

Utilisez les exécutions async dans les serveurs web. Utilisez les exécutions sync pour les scripts et les tâches planifiées.

Gérer les erreurs

L’intégration remonte les erreurs HTTP du client TwexAPI. Branchez sur le statut avant de réessayer.

Enregistrez les codes de statut, jamais les identifiants. Limitez les retries pour éviter des boucles d’agent non bornées.

Composant Haystack ou API REST directe

Votre pipeline renvoie déjà des objets Document ? Ajoutez ces composants directement. Ils normalisent le texte de tweet, les métadonnées, les URLs et la pagination.

Utilisez les routes REST TwexAPI directes ou le SDK Python pour les abonnés, following, réponses, citations, listes, communautés, médias, tendances et écritures approuvées.

Les deux approches utilisent les mêmes contrats TwexAPI. Les composants ne supportent que la recherche de tweets et les timelines utilisateur.

Alternative de handoff MCP

Lorsqu’un agent découvre d’abord les endpoints, utilisez Twexapi MCP pour la collecte et convertissez les lignes en objets Haystack Document manuellement. Préférez les composants Haystack lorsque le pipeline s’exécute déjà dans Haystack et que vous voulez une pagination typée sans découverte par agent.

Source et contrats

• Dépôt x-api-scraper-haystack
• Release Haystack 3.0
• Advanced Twitter Search
• User Timeline
• Agent MCP Handoff