커다란 건초 더미

TwexAPI 트윗 검색 및 사용자 타임라인 구성 요소, 입력된 문서, 인용 및 페이지 매기기를 사용하여 Haystack RAG 파이프라인을 구축하세요.

Haystack은 Python AI 애플리케이션을 위한 오픈 소스 프레임워크입니다. RAG 파이프라인 및 에이전트 워크플로의 현재 트윗을 보려면 x-api-scraper-haystack을 사용하세요. 통합은 읽기 전용이며 쓰기 액세스 권한을 부여하지 않습니다.

통합은 두 가지 읽기 전용 구성 요소를 제공합니다.

검색, 타임라인, 모니터링 및 검색 증강 생성에 이러한 구성 요소를 사용합니다. 에이전트에 검색 도구가 필요한 경우 TwexApiTweetSearch를 ComponentTool로 래핑하세요.

팔로어 내보내기에는 팔로어 API를 사용하세요. 승인된 게시를 위해 쓰기 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"]

최근 모니터링에는 ’최신’을 사용하세요. 참여 순위 검색에는 ’상위’를 사용하세요. 순위는 바뀔 수 있으므로 트윗 ID를 저장하세요.

트윗 ID 옆에 모든 검색어를 저장하세요. 각 실행의 트윗 수를 제한하려면 top_k를 설정하세요.

집중적인 트윗 검색 구축

전체 쿼리 옵션은 고급 Twitter 검색을 참조하세요. 실행 시 타임스탬프와 커서를 전달할 때 구성 요소 초기화 외부에 유지합니다.

Twitter 사용자 타임라인 가져오기

POST /twitter/{screen_name}/timeline/page에는 TwexApiUserTweetsFetcher를 사용하세요. 화면 이름을 전달하세요.

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

여러 계정 검색 및 하나의 계정에 대한 타임라인 검색을 선택하세요.

Haystack 문서 필드 이해

각 트윗은 하나의 Haystack ’Document’가 됩니다. 트윗 텍스트는 Document.content가 됩니다. 안정적인 필드는 메타데이터가 됩니다.

누락된 필드는 그대로 유지됩니다. 누락된 측정항목을 0으로 처리하지 마세요. 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은 하나의 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())

웹 서버에서 비동기 실행을 사용합니다. 스크립트 및 예약된 작업에 대해 동기화 실행을 사용합니다.

오류 처리

통합하면 TwexAPI 클라이언트에서 HTTP 오류가 발생합니다. 재시도하기 전에 상태에 따라 분기하세요.

자격 증명이 아닌 상태 코드를 기록하십시오. 제한되지 않은 에이전트 루프를 방지하기 위해 Cap 재시도가 이루어집니다.

Haystack 구성 요소 또는 직접 REST API

파이프라인이 이미 Document 개체를 반환합니까? 이러한 구성요소를 직접 추가하세요. 트윗 텍스트, 메타데이터, URL 및 페이지 매김을 표준화합니다.

팔로어, 팔로우, 답글, 인용문, 목록, 커뮤니티, 미디어, 트렌드 및 승인된 쓰기에 대해 직접 TwexAPI REST 경로 또는 Python SDK를 사용하세요.

두 접근 방식 모두 동일한 TwexAPI 계약을 사용합니다. 구성 요소는 트윗 검색 및 사용자 타임라인만 지원합니다.

MCP 핸드오프 대안

에이전트가 엔드포인트를 먼저 발견하면 수집을 위해 Twexapi MCP를 사용하고 행을 수동으로 Haystack ‘Document’ 개체로 변환합니다. 파이프라인이 이미 Haystack 내에서 실행되고 있고 에이전트 검색 없이 입력된 페이지 매김을 원하는 경우 Haystack 구성 요소를 선호하세요.

소스 및 계약

• x-api-scraper-haystack 저장소
• Haystack 3.0 릴리스
• 고급 Twitter 검색
• 사용자 타임라인
• 에이전트 MCP 핸드오프