用 Airbyte 搭建 AI 应用数据管道

AI 应用接上模型之后，麻烦往往才开始：外部数据怎么拉、分页怎么翻、接口变了谁来修？作者在 Career Agency AI 项目里，需要汇集招聘信息，再交给 Gemini 分析。他的选择是把采集和同步交给 Airbyte，让 FastAPI 负责读取数据和调用模型。

这套分工值得参考，但原文更像一份架构演示。它说明了数据如何流动，却没有证明清洗、结构变化和持续运行的问题都已解决。

为什么先把数据同步从业务接口里拆出来

作者最初为 Remotive、Arbeitnow 等招聘来源分别编写 Python 脚本，手动处理分页、请求频率限制和数据结构变化。接入来源越多，需要维护的接口细节也越多。

换用 Airbyte 后，链路变成：

外部招聘 HTTP API → Airbyte 同步 → 本地 PostgreSQL → FastAPI → Gemini API。

这里最有用的变化，是业务接口可以读取已经落库的数据，不必在用户请求到来时，再逐个访问招聘来源。采集任务与用户请求分开，故障也更容易定位。

原文称它为 ETL，即抽取、转换、加载。不过，展示出来的主要是抽取和加载，清洗、字段统一、去重等转换步骤没有展开。作者也提到移动部署可用 SQLite，但没有提供对应配置，实际演示围绕 PostgreSQL。

Airbyte 接管了什么，又有哪些配置没交代

作者采用 Docker 部署，并给出以下启动命令：

git clone https://github.com/airbytehq/airbyte.git
cd airbyte
docker compose up

要留意最后一行：这是原文提供的启动方式，但没有锁定 Airbyte 版本，也没有说明所需 Compose 文件和运行条件，不能据此保证任意版本都能直接启动。

原文给出的界面地址是 http://localhost:8000。进入后，作者通过 HTTP API connector 配置数据源；接口比较特殊时，则使用自定义 Python connector。目标端选择 PostgreSQL。

作者用「几次点击替代 200 行 Python 脚本」描述收益。这个方向合理：通用同步工作可以交给工具。但原文没有展示连接器配置、分页规则、同步周期、增量字段或失败重试策略，读者还无法照着完整复现。

Airbyte 能减少重复接入代码，但连接器仍然需要理解源接口。

FastAPI 到 Gemini 的最后一段还缺什么

数据库同步完成后，作者用 psycopg2.connect 连接数据库：dbname="career_agency"、user="postgres"、password="[YOUR_PASSWORD]"、host="localhost"、port="5432"。

GET /jobs/latest 执行的查询是：

SELECT title, company, description FROM job_listings ORDER BY created_at DESC LIMIT 10;

接口取最新 10 条记录，查不到则通过 HTTPException 返回 404，详情为 No jobs found。这里隐含了一个前提：同步结果中已经存在 job_listings 表，以及查询所需的字段。原文没有给出建表或字段映射过程，不能默认连接器会生成完全相同的结构。

生成求职信的示例展示了模型调用位置：

import google.generativeai as genai

genai.configure(api_key="[YOUR_GEMINI_KEY]")

@app.post("/generate-cover-letter")
def generate_letter(job_id: int):
    # 1. Fetch job description from the Airbyte-synced DB
    # 2. Pass to Gemini
    model = genai.GenerativeModel('gemini-pro')
    response = model.generate_content(f"Write a cover letter for this job: {job_description}")

    return {"cover_letter": response.text}

要看的是 generate_content 那一行：它依赖的 job_description 没有定义，前面的两条注释也没有实现查询。代码说明了调用思路，不能直接运行。

另外，job_id: int 按这段 FastAPI 声明会作为查询参数接收，并不是 JSON 请求体。前端对接时需要明确这个约定。原文使用 google.generativeai 和 gemini-pro，没有提供 SDK 版本或兼容性说明；应把它们视为原文示例，而非已经核验的当前选型。

我的判断

我认可这种职责划分：Airbyte 管接入和同步，数据库保存可查询的数据，FastAPI 承接业务，Gemini 负责生成内容。对于多个来源、需要持续更新的应用，这比让采集逻辑散落在业务接口中更容易维护。

但作者声称「数据始终新鲜」「结构变化自动处理」「节省数小时」，没有给出同步延迟、故障记录或前后对照数据。这些属于项目经验，不能直接推广成工具保证。字段能够同步，也不代表字段改名后，业务查询还能正常执行。

作者还说自己硬件受限，因此选择 Docker 部署，却没有提供内存、CPU 或容器资源占用。Docker 是部署方式，这段经历不足以证明 Airbyte 适合低配机器。

如果只有一两个稳定来源，定时脚本加数据库也可能更省事。来源增加、同步需求重复时，Airbyte 的价值才更容易显现；代价是团队需要维护额外的同步服务。

能拿来做什么

• 为多来源 AI 应用先画清数据链路，把采集任务与 FastAPI 或 Node 业务接口分开，让用户请求读取已经落库的数据。
• 接前端之前，先核对同步后的真实表名、字段和返回格式，再实现 /jobs/latest，不要直接假定数据库符合示例。
• 补全 /generate-cover-letter 的数据库查询，明确 job_id 的传参方式，并核验所用 Gemini SDK 与模型是否可用。
• 用同一个招聘来源比较定时脚本与 Airbyte 的配置、维护成本，再决定是否扩展到其他来源。