> ## Agent Instructions
>
> Base URL: https://api.anysite.io
> Authentication: send the `access-token` header. Do NOT use `Authorization: Bearer`.
> Full endpoint catalog: https://app.anysite.io/docs
# 源类型

用于构建灵活数据采集管道的六种数据源类型

## 概述

数据集管道支持 6 种数据源类型，每种都对应不同的数据采集模式。多种数据源可以组合起来构建复杂的多步骤工作流。
下面的独立型（Independent）、文件型（From File）和依赖型（Dependent）本质上都是同一种底层
`type: api` 数据源（省略 `type` 时的默认值）— 它们的区别仅在于设置了哪些字段。

| 类型 | 用途 | 关键配置 |
|------|---------|------------|
| **独立型（Independent）** | 使用静态参数的单次 API 调用 | `endpoint`、`params` |
| **文件型（From File）** | 遍历文件进行批量调用 | `from_file`、`input_key` |
| **依赖型（Dependent）** | 使用父数据源的值进行批量调用 | `dependency`、`input_key` |
| **合并型（Union）** | 合并多个数据源的记录 | `type: union`、`sources` |
| **LLM** | 通过 LLM 模型处理数据 | `type: llm`、`llm` |
| **SQL** | 查询 Parquet 视图或指定的数据库连接 | `type: sql`、`query` |

## 独立型数据源（Independent Source）

```yaml
sources:
  - id: search_results
    endpoint: /api/linkedin/search/users
    params:
      keywords: "CTO"
      count: 50
    parallel: 1
    rate_limit: "10/s"
    on_error: stop
```

**适用场景：** 管道的起点、搜索查询、单个资料查询。

## 文件型数据源（From File Source）

```yaml
sources:
  - id: company_profiles
    endpoint: /api/linkedin/company
    from_file: companies.txt
    input_key: company
    parallel: 3
    rate_limit: "10/s"
    on_error: skip
```

支持的文件格式：
- **TXT** — 每行一个值
- **CSV** — 使用与 `input_key` 匹配的列
- **JSON/JSONL** — 使用与 `input_key` 匹配的字段

**适用场景：** 您已经有一份 URL、ID 或搜索词的列表。

## 依赖型数据源（Dependent Source）

```yaml
sources:
  - id: companies
    endpoint: /api/linkedin/search/companies
    params:
      keywords: "AI startup"
      count: 100

  - id: employees
    endpoint: /api/linkedin/company/employees
    dependency:
      from_source: companies
      field: urn.value
      dedupe: true
    input_key: companies
    parallel: 3
    rate_limit: "10/s"
    on_error: skip
```

### 依赖配置

| 字段 | 描述 |
|-------|-------------|
| `from_source` | 父数据源的 ID |
| `field` | 从父数据源结果中提取的字段路径（支持点号表示法） |
| `dedupe` | 处理前去除重复值（默认：`false`） |

**简写形式：** 对于单个依赖参数，在 `params` 中内联使用 `${source.field}` 会自动展开为等价的
`dependency`/`input_key` 组合：

```yaml
  - id: employees
    endpoint: /api/linkedin/company/employees
    params:
      companies: "${companies.urn.value}"
```

等价于上面显式的 `dependency: {from_source: companies, field: urn.value}` /
`input_key: companies` 形式。同一个数据源上不能混用这两种写法。

支持**多级链式依赖** — 一个依赖型数据源本身也可以作为另一个依赖型数据源的父数据源：

```yaml
sources:
  - id: companies
    endpoint: /api/linkedin/search/companies
    params: { keywords: "AI", count: 50 }

  - id: employees
    endpoint: /api/linkedin/company/employees
    dependency: { from_source: companies, field: urn.value }
    input_key: companies

  - id: profiles
    endpoint: /api/linkedin/user
    dependency: { from_source: employees, field: urn.value }
    input_key: user

  - id: posts
    endpoint: /api/linkedin/user/posts
    dependency: { from_source: profiles, field: urn.value }
    input_key: user
```

**适用场景：** 多步骤数据增强，从搜索结果到详细资料再到活动数据。

## 合并型数据源（Union Source）

```yaml
sources:
  - id: search_cto
    endpoint: /api/linkedin/search/users
    params: { keywords: "CTO", count: 50 }

  - id: search_vp
    endpoint: /api/linkedin/search/users
    params: { keywords: "VP Engineering", count: 50 }

  - id: all_leaders
    type: union
    sources: [search_cto, search_vp]
    dedupe_by: urn.value
```

### 合并配置

| 字段 | 描述 |
|-------|-------------|
| `type` | 必须为 `union` |
| `sources` | 要合并的数据源 ID 列表（必须共用同一个端点） |
| `dedupe_by` | 用于去重的字段（可选） |

**适用场景：** 合并多次搜索的结果，整合来自不同平台的数据。

## LLM 数据源（LLM Source）

```yaml
sources:
  - id: profiles
    endpoint: /api/linkedin/user
    from_file: users.txt
    input_key: user

  - id: profiles_analyzed
    type: llm
    dependency:
      from_source: profiles
      field: name
    llm:
      - type: classify
        categories: "developer,recruiter,executive,other"
        output_column: role_type

      - type: enrich
        add:
          - "seniority:junior/mid/senior/lead"
          - "is_technical:boolean"

      - type: summarize
        max_length: 50
        output_column: bio_summary
```

### LLM 操作

| 操作 | 描述 |
|-----------|-------------|
| `classify` | 将记录分类到预定义类别 |
| `enrich` | 提取新属性（枚举、字符串、布尔值、数字） |
| `summarize` | 生成简洁摘要 |
| `generate` | 使用带字段占位符的模板生成文本 |

> 
  管道中的 LLM 步骤仅支持这 4 种操作 — `match` 和 `deduplicate` 只能作为独立的
  [`anysite llm match` / `anysite llm deduplicate`](/docs/cli/llm-analysis?lang=zh) 命令使用，
  不能作为 `type: llm` 管道步骤使用。需要 `llm` 可选依赖包：`pip install "anysite-cli[llm]"`。

**适用场景：** 为管道添加 AI 驱动的增强能力，对采集到的数据进行分类或摘要。

## SQL 数据源（SQL Source）

运行一次 SQL 查询，并将结果存储为它自己的数据源 — 既可以针对同一管道中已采集数据源的 Parquet 视图，
也可以针对一个指定的数据库连接：

```yaml
sources:
  # Query dataset Parquet files via DuckDB (no external DB needed)
  - id: enriched_profiles
    type: sql
    query: |
      SELECT u.*, c.description as company_desc
      FROM profiles u
      LEFT JOIN companies c ON u._input_value = c._input_value

  # Query an external database via a named connection ('anysite db add')
  - id: billing_data
    type: sql
    connection: pg
    query: "SELECT name, email FROM subscriptions"
```

可以用 `query_file`（指向 `.sql` 文件的路径）代替内联的 `query`。省略 `connection` 表示查询该管道自身已
采集的数据源。

## Per-Source Transform & Export

```yaml
sources:
  - id: companies
    endpoint: /api/linkedin/company
    from_file: companies.txt
    input_key: company
    transform:
      filter: '.employee_count > 10'
      fields: [name, url, employee_count]
      add_columns:
        batch: "q1-2026"
    export:
      - type: file
        path: ./output/companies-{{date}}.csv
        format: csv
    db_load:
      key: _input_value
      sync: full
      fields: [name, url, employee_count]
```

### 转换选项

| 字段 | 描述 |
|-------|-------------|
| `filter` | jq 风格的过滤表达式，保留匹配的记录 |
| `fields` | 要包含在输出中的字段列表 |
| `add_columns` | 为每条记录添加的静态列 |

### 导出选项

| 字段 | 描述 |
|-------|-------------|
| `type` | 导出类型：`file` 或 `webhook` |
| `path` | 输出文件路径（`file` 类型；支持 `{{date}}` 模板） |
| `format` | 导出格式：`csv`、`json`、`jsonl`（默认：`jsonl`） |
| `url` | Webhook URL（`webhook` 类型） |
| `headers` | Webhook 的额外 HTTP 头 |

### 数据库加载选项（按数据源）

| 字段 | 描述 |
|-------|-------------|
| `key` | 用于增量同步的唯一键列 |
| `sync` | 同步模式：`full`（默认，包含 DELETE）或 `append`（不执行 DELETE） |
| `fields` | 要加载到数据库中的字段 |

## 输入模板

```yaml
sources:
  - id: employees
    endpoint: /api/linkedin/company/employees
    dependency:
      from_source: companies
      field: urn.value
    input_key: companies
    input_template:
      companies:
        - type: company
          value: "{value}"
      count: 5
```

`{value}` 占位符会被依赖关系中的每个输入值替换。

## 通用数据源选项

以下选项适用于 `type: api` 数据源（独立型、文件型、依赖型）：

| 选项 | 描述 | 默认值 |
|--------|-------------|---------|
| `parallel` | 并发工作进程数量 | `1` |
| `rate_limit` | 最大请求速率（例如 `"10/s"`） | 无限制 |
| `on_error` | 错误处理方式：`stop` 或 `skip`（数据源配置层级没有 `retry`） | `skip` |
| `refresh` | 增量行为：`auto`、`always` | `auto` |

## 后续步骤

  
#### Scheduling

    设置增量采集、cron/systemd 条目和 Webhook
  
  
#### Database Loading

    将管道结果加载到 SQLite、PostgreSQL 或 ClickHouse