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

Haystack 是 Python AI 应用的开源框架。在 RAG 管道与Agent 工作流中获取最新推文请用 [`x-api-scraper-haystack`](https://github.com/twexapi-dev/x-api-scraper-haystack)。该集成为只读，永不授予写权限。

集成提供两个只读组件：

<CardGroup cols={2}>
  <Card title="搜索公开推文" icon="search">
    `TwexApiTweetSearch` 通过 `POST /twitter/advanced_search/page` 搜索关键词、话题标签、账号与 X 查询运算符。
  </Card>

  <Card title="拉取用户时间线" icon="user">
    `TwexApiUserTweetsFetcher` 通过 `POST /twitter/{screen_name}/timeline/page` 获取单个公开账号时间线。
  </Card>
</CardGroup>

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

粉丝导出用 [followers API](/api-reference/followers-following-endpoints/followers-v3-api-v3-twitter-users-followers-post)。经批准的发布用 [write API](/api-reference/tweet-actions-endpoints/create-tweet-twitter-tweets-create-post)。这些操作不在本集成内。

## 安装 Haystack 与 TwexAPI

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

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

在虚拟环境中安装。

在 [TwexAPI dashboard](https://twexapi.io/dashboard) 创建 API 密钥，本地导出：

```bash
export X_API_SCRAPER_KEY="YOUR_API_KEY"
```

切勿将生产密钥嵌入管道 YAML、笔记本或源码控制。通过 Haystack `Secret` 对象加载环境变量。

## 在 Python 中搜索 Twitter 推文

`TwexApiTweetSearch` 调用 `POST /twitter/advanced_search/page`，接受 X 搜索语法并返回 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"]
```

近期监控用 `Latest`；按互动发现用 `Top`。排名可能变化，请保存 tweet ID。

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

### 构建聚焦推文搜索

| 搜索意图   | 查询示例                                |
| --------------- | -------------------------------------------- |
| 精确短语    | `"retrieval augmented generation"`           |
| 账号发帖   | `from:deepset_ai haystack`                   |
| 话题标签  | `#haystack #rag`                             |
| 日期窗口     | `haystack since:2026-07-01 until:2026-08-01` |
| 排除转推 | `haystack -filter:retweets`                  |

完整查询选项见 [Advanced Twitter Search](/api-reference/search-endpoints/get-data-page-twitter-advanced-search-page-post)。按运行传入时间戳与游标时，勿在组件 init 中写死。

## 获取 Twitter 用户时间线

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

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

多账号用搜索，单账号用时间线。

## 理解 Haystack document 字段

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

| Document 字段                                               | 存储的推文值                           |
| ------------------------------------------------------------ | -------------------------------------------- |
| `content`                                                    | 来自 `full_text` 或 `text` 的推文文本        |
| `meta.endpoint`                                              | `search` 或 `timeline`                       |
| `meta.id`, `meta.url`                                        | 可用时的 tweet ID 与规范 URL    |
| `meta.created_at`                                            | 时间戳                                    |
| `meta.author`                                                | 作者 ID、用户名、名称与 verified 标志   |
| `meta.like_count`, `meta.retweet_count`, `meta.reply_count`  | 点赞、转推与回复                  |
| `meta.quote_count`, `meta.view_count`, `meta.bookmark_count` | 可用时的引用、浏览与书签  |

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

## 在 RAG 引用中保留证据

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

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

引用须使用提供的 URL。拒绝检索文档中不存在的 URL。

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

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

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

## 索引推文或实时检索

<CardGroup cols={2}>
  <Card title="实时在线检索" icon="radio">
    每个问题运行时搜索，获取最新推文与活跃事件。
  </Card>

  <Card title="本地索引语料库" icon="database">
    在稳定时间窗口内重复研究时存储 embedding。
  </Card>
</CardGroup>

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

## 分页且不重复推文

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

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

每页文档后保存游标。切勿编辑游标。按 `Document.meta["id"]` 去重。

## 异步运行 Haystack 管道

Haystack 3 使用单一 `Pipeline` 类。两 TwexAPI 组件均暴露 `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())
```

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

## 处理错误

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

| 状态 | 含义                       | 操作                            |
| ------ | ----------------------------- | --------------------------------- |
| `400`  | 无效请求               | 修正。**不要**原样重试。   |
| `401`  | API 密钥无效               | 添加有效 API 密钥。              |
| `403`  | 访问被拒                 | 检查账号访问与额度。 |
| `429`  | 超出速率限制           | 重试前退避。         |
| `5xx`  | 临时服务错误        | 有界退避重试。       |

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

## Haystack 组件或直接 REST API

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

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

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

## MCP handoff 替代方案

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

## 源码与契约

* [x-api-scraper-haystack repository](https://github.com/twexapi-dev/x-api-scraper-haystack)
* [Haystack 3.0 release](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)