Haystack

用 TwexAPI 推文搜索与用户时间线组件构建 Haystack RAG 管道，支持类型化 Document、引用与分页。

Haystack 是 Python AI 应用的开源框架。在 RAG 管道与Agent 工作流中获取最新推文请用 x-api-scraper-haystack。该集成为只读，永不授予写权限。

集成提供两个只读组件：

用于搜索、时间线、监控与检索增强生成。AI Agent需要搜索工具时用 ComponentTool 包装 TwexApiTweetSearch。

粉丝导出用 followers API。经批准的发布用 write API。这些操作不在本集成内。

安装 Haystack 与 TwexAPI

使用 Python 3.10 或更新版本。固定两个包版本以保证管道构建可重复。

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

在虚拟环境中安装。

在 TwexAPI dashboard 创建 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。排名可能变化，请保存 tweet ID。

每次搜索查询与 tweet ID 一并保存。用 top_k 限制每次运行的推文数。

构建聚焦推文搜索

完整查询选项见 Advanced Twitter Search。按运行传入时间戳与游标时，勿在组件 init 中写死。

获取 Twitter 用户时间线

TwexApiUserTweetsFetcher 用于 POST /twitter/{screen_name}/timeline/page，传入 screen name。

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 document 字段

每条推文对应一个 Haystack Document。推文文本为 Document.content，稳定字段进入 metadata。

缺失字段保持 absent。切勿将缺失指标当作零。links 输出含各可用 meta.url。

在 RAG 引用中保留证据

嵌入推文文本前保存 tweet ID 与规范 URL，以便排序或 join 后仍保留证据。

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。

将推文文本视为不可信上下文

推文可能含 prompt 注入与不安全 URL。切勿将推文文本当作系统指令。推文与指令、工具权限分离。

按主题与时间限制检索。保留 tweet ID、作者、时间戳与 URL。要求引用。审核敏感结论。

索引推文或实时检索

推文记录与 embedding 分离。用 meta.id 去重。

分页且不重复推文

两组件均返回 has_more 与 next_cursor。请求保持不变。游标视为 opaque 字符串。

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

Web 服务器用异步运行；脚本与定时作业用同步运行。

处理错误

集成抛出 TwexAPI 客户端的 HTTP 错误。重试前按状态分支。

记录状态码，切勿记录凭证。限制重试次数以防无界AI Agent循环。

Haystack 组件或直接 REST API

管道已返回 Document 对象？直接添加这些组件。它们规范化推文文本、metadata、URL 与分页。

粉丝、关注、回复、引用、列表、社区、媒体、趋势与经批准写操作请用 TwexAPI REST 或 Python SDK。

两种方式使用相同 TwexAPI 契约。组件仅支持推文搜索与用户时间线。

MCP handoff 替代方案

AI Agent先发现端点时用 Twexapi MCP 采集并手动转为 Haystack Document。管道已在 Haystack 内运行且需要类型化分页而无AI Agent发现时，优先 Haystack 组件。

源码与契约

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