--- title: "Haystack" description: "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`](https://github.com/twexapi-dev/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 : `TwexApiTweetSearch` recherche des mots-clés, hashtags, comptes et opérateurs de requête X via `POST /twitter/advanced_search/page`. `TwexApiUserTweetsFetcher` récupère la timeline d'un compte public via `POST /twitter/{screen_name}/timeline/page`. 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](/api-reference/followers-following-endpoints/followers-v3-api-v3-twitter-users-followers-post) pour les exports d'abonnés. Utilisez l'[API d'écriture](/api-reference/tweet-actions-endpoints/create-tweet-twitter-tweets-create-post) 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. ```bash python -m pip install "x-api-scraper-haystack" "haystack-ai>=3.0.0" ``` Installez dans un environnement virtuel. Créez une [clé API TwexAPI](https://twexapi.io/dashboard), puis exportez-la localement. ```bash 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`. ```python 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 | Intention de recherche | Exemple de requête | | ---------------------- | --------------------------------------------- | | Phrase exacte | `"retrieval augmented generation"` | | Publications d'un compte | `from:deepset_ai haystack` | | Recherche par hashtag | `#haystack #rag` | | Plage de dates | `haystack since:2026-07-01 until:2026-08-01` | | Exclure les republications | `haystack -filter:retweets` | Consultez [Advanced Twitter Search](/api-reference/search-endpoints/get-data-page-twitter-advanced-search-page-post) 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. ```python 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. | Champ Document | Valeur tweet stockée | | ------------------------------------------------------------ | ---------------------------------------------- | | `content` | Texte du tweet depuis `full_text` ou `text` | | `meta.endpoint` | `search` ou `timeline` | | `meta.id`, `meta.url` | ID de tweet et URL canonique si disponible | | `meta.created_at` | Horodatage | | `meta.author` | ID auteur, username, nom et flag vérifié | | `meta.like_count`, `meta.retweet_count`, `meta.reply_count` | Likes, republications et réponses | | `meta.quote_count`, `meta.view_count`, `meta.bookmark_count` | Citations, vues et bookmarks si disponibles | 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. ```python 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 Recherchez à chaque question pour des tweets récents et des événements actifs. Stockez les embeddings pour des recherches répétées sur des fenêtres stables. 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. ```python 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()`. ```python 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. | Statut | Signification | Action | | ------ | ----------------------------- | --------------------------------- | | `400` | Requête invalide | Corrigez. Ne pas réessayer sans modification. | | `401` | Clé API invalide | Ajoutez une clé API valide. | | `403` | Accès refusé | Vérifiez l'accès au compte et les crédits. | | `429` | Limite de débit dépassée | Ralentissez avant de réessayer. | | `5xx` | Erreur serveur transitoire | Réessayez avec backoff borné. | 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](/sdks/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](/mcp/overview) 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](https://github.com/twexapi-dev/x-api-scraper-haystack) * [Release Haystack 3.0](https://github.com/deepset-ai/haystack/releases/tag/v3.0.0) * [Advanced Twitter Search](/api-reference/search-endpoints/get-data-page-twitter-advanced-search-page-post) * [User Timeline](/api-reference/timeline-endpoints/get-user-timeline-page-api-twitter-screen-name-timeline-page-post) * [Agent MCP Handoff](/mcp/agent-handoff)