Haystack

Construye pipelines RAG con Haystack usando componentes de búsqueda de tweets y líneas de tiempo de TwexAPI, Documents tipados, citas y paginación.

Haystack es un framework de código abierto para aplicaciones de IA en Python. Usa x-api-scraper-haystack para tweets actuales en pipelines RAG y flujos de trabajo de agentes. La integración es de solo lectura y nunca otorga acceso de escritura.

La integración proporciona dos componentes de solo lectura:

Usa estos componentes para búsqueda, líneas de tiempo, monitoreo y generación aumentada por recuperación. Envuelve TwexApiTweetSearch con ComponentTool cuando un agente necesite una herramienta de búsqueda.

Usa la API de seguidores para exportaciones de seguidores. Usa la API de escritura para publicación aprobada. Esas acciones quedan fuera de esta integración.

Instalar Haystack y TwexAPI

Usa Python 3.10 o más reciente. Fija ambos paquetes para builds de pipeline reproducibles.

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

Instala dentro de un entorno virtual.

Crea una API key de TwexAPI, luego expórtala localmente.

export X_API_SCRAPER_KEY="YOUR_API_KEY"

Nunca incrustes API keys de producción en YAML de pipeline, notebooks o control de versiones. Carga la variable de entorno a través de un objeto Secret de Haystack.

Buscar tweets de Twitter en Python

TwexApiTweetSearch llama a POST /twitter/advanced_search/page. Acepta sintaxis de búsqueda de X y devuelve objetos Document de Haystack.

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

Usa Latest para monitoreo reciente. Usa Top para descubrimiento ordenado por engagement. Almacena IDs de tweet porque los rankings pueden cambiar.

Guarda cada consulta de búsqueda junto a sus IDs de tweet. Establece top_k para limitar la cantidad de tweets en cada ejecución.

Construir búsquedas de tweets enfocadas

Consulta Advanced Twitter Search para opciones completas de consulta. Mantén marcas de tiempo y cursores fuera de la inicialización del componente cuando los pases por ejecución.

Obtener la línea de tiempo de un usuario de Twitter

Usa TwexApiUserTweetsFetcher para POST /twitter/{screen_name}/timeline/page. Pasa un nombre de pantalla.

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

Elige búsqueda para muchas cuentas y líneas de tiempo para una cuenta.

Entender los campos de documento de Haystack

Cada tweet se convierte en un Document de Haystack. El texto del tweet se convierte en Document.content. Los campos estables se convierten en metadatos.

Los campos faltantes permanecen ausentes. Nunca trates métricas faltantes como cero. La salida links contiene cada meta.url disponible.

Mantener evidencia en citas RAG

Almacena IDs de tweet y URLs canónicas antes de incrustar el texto del tweet. Esto preserva la evidencia después del ranking o uniones.

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

Exige URLs proporcionadas para citas. Rechaza URLs ausentes de los documentos recuperados.

Tratar el texto del tweet como contexto no confiable

Los tweets pueden contener inyección de prompts y URLs inseguras. Nunca trates el texto del tweet como una instrucción del sistema. Mantén los tweets separados de instrucciones y permisos de herramientas.

Limita la recuperación por tema y tiempo. Conserva IDs de tweet, autores, marcas de tiempo y URLs. Exige citas. Revisa conclusiones sensibles.

Indexar tweets u obtenerlos en vivo

Mantén los registros de tweets separados de los embeddings. Usa meta.id para deduplicación.

Paginar sin tweets duplicados

Ambos componentes devuelven has_more y next_cursor. Mantén la solicitud sin cambios. Trata los cursores como cadenas opacas.

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

Guarda cada cursor después de sus documentos. Nunca edites un cursor. Deduplica por Document.meta["id"].

Ejecutar pipelines de Haystack de forma asíncrona

Haystack 3 usa una clase Pipeline. Ambos componentes de TwexAPI exponen 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())

Usa ejecuciones async en servidores web. Usa ejecuciones sync para scripts y trabajos programados.

Manejar errores

La integración lanza errores HTTP del cliente TwexAPI. Ramifica por estado antes de reintentar.

Registra códigos de estado, nunca credenciales. Limita reintentos para evitar bucles de agente sin límite.

Componente de Haystack o API REST directa

¿Tu pipeline ya devuelve objetos Document? Añade estos componentes directamente. Normalizan texto de tweet, metadatos, URLs y paginación.

Usa rutas REST directas de TwexAPI o el Python SDK para seguidores, following, respuestas, citas, listas, comunidades, medios, tendencias y escrituras aprobadas.

Ambos enfoques usan los mismos contratos de TwexAPI. Los componentes solo admiten búsqueda de tweets y líneas de tiempo de usuario.

Alternativa de entrega MCP

Cuando un agente descubre endpoints primero, usa Twexapi MCP para recopilación y convierte las filas en objetos Document de Haystack manualmente. Prefiere los componentes de Haystack cuando el pipeline ya corre dentro de Haystack y quieres paginación tipada sin descubrimiento del agente.

Fuente y contratos

• Repositorio x-api-scraper-haystack
• Lanzamiento de Haystack 3.0
• Advanced Twitter Search
• User Timeline
• Agent MCP Handoff