用 CrewAI 与 BrightData 构建 YouTube 趋势分析系统：从频道采集到多智能体报告生成的完整实现

本文以 YouTube 趋势分析项目 为主体，完整讲解该项目如何用 BrightData 采集指定 YouTube 频道的视频及字幕、用 CrewAI 双智能体流水线分析字幕文本并生成趋势报告、用 Streamlit 搭建可交互的 Web 界面。读完后你将掌握：三组件（数据采集层 / 多智能体分析层 / 展示层）的职责划分与调用链路、BrightData Datasets v3 API 的触发-轮询-取结果三步调用方式、CrewAI Agent/Task 的 YAML 化配置，以及从 streamlit run app.py 到拿到 youtube_trend_analysis.md 报告的全部操作步骤。

一、项目总览：一条"采集 → 落盘 → 分析 → 呈现"的数据流水线

项目在 README 中给出了三行核心定位，正好对应其三层架构：

• BrightData：负责从 YouTube 频道抓取视频（含视频链接与完整字幕 formatted_transcript）；
• CrewAI：负责分析抓取到的视频字幕文本，生成结构化趋势总结；
• Streamlit：提供 Web 交互界面，串联采集与分析两个阶段。

从 入口文件 的源码结构看，一次完整分析的调用链如下：

1. 用户在侧边栏输入一个或多个频道 URL 与日期范围，点击 "Start Analysis"，触发 start_analysis()（app.py#L100）；
2. 调用 trigger_scraping_channels() 向 BrightData 发起异步采集任务，然后每 10 秒轮询一次 get_progress()，直到快照状态变为 ready；
3. 调用 get_output() 拉取 JSON 结果，将每条视频的 formatted_transcript 按时间轴写入 transcripts/<video_id>.txt；
4. 用 config.yaml 中定义的两个 Agent 与两个 Task 构建 Crew，以 file_paths（全部字幕文件路径）为输入执行 crew.kickoff()；
5. 最终报告以 Markdown 渲染在页面中，并可通过下载按钮保存为 youtube_trend_analysis.md（app.py#L247-L260）。

模块分工对应的仓库文件为：

• app.py：Streamlit 界面 + CrewAI 编排 + 字幕落盘；
• brightdata_scrapper.py：封装 BrightData Datasets v3 API 的四个函数；
• config.yaml：Agent 与 Task 的角色、目标、背景设定及任务描述/期望输出。

二、环境准备：API Key、Ollama 与依赖安装

以下三部分与 README 中的 Setup 章节一一对应。

2.1 配置 BrightData API Key

在 BrightData 控制台注册并创建 API Key 后，在项目根目录创建 .env 文件：

BRIGHT_DATA_API_KEY=your_api_key

应用通过 load_dotenv() 读取该文件：app.py#L11-L12 与 brightdata_scrapper.py#L5-L8 均在模块加载时执行 load_dotenv()，随后用 os.getenv("BRIGHT_DATA_API_KEY") 取值。

此外，app.py#L24 中还通过 os.getenv("OPENAI_API_KEY") 读取 OpenAI Key 作为默认 LLM 的凭据（详见第五节的 LLM 配置说明），如选用 gpt-4o，需要在 .env 中同时提供该变量。

2.2 安装 Ollama 并拉取本地模型

README 提供了本地模型路线的准备工作（适用于希望不用付费 API 的场景）：

# setup ollama on linux
curl -fsSL https://ollama.com/install.sh | sh
# pull llama 3.2 model
ollama pull llama3.2

需要注意一个实现细节：app.py#L21-L30 中 load_llm() 当前生效的配置是 LLM(model="gpt-4o", api_key=os.getenv("OPENAI_API_KEY"))，而 Ollama 本地模型是一行被注释掉的备选：

llm = LLM(model="ollama/llama3.2", base_url="http://localhost:11434")

也就是说，README 描述的 Ollama 路线对应的是把注释行取消注释、注释掉 gpt-4o 一行。@st.cache_resource 装饰器保证该 LLM 实例在整个 Streamlit 会话中只创建一次，避免反复初始化。

2.3 安装 Python 依赖

要求 Python 3.11 或更高版本：

pip install streamlit ollama crewai crewai-tools

从 app.py 的实际导入看，运行时还依赖 python-dotenv（load_dotenv）、pyyaml（解析 config.yaml）与 tqdm（字幕处理的进度条），它们是上述依赖的常见传递依赖，若环境缺失可单独补装。

三、运行项目

一切就绪后，在项目根目录执行：

streamlit run app.py

界面提供三部分交互能力（见 app.py#L193-L239 的侧边栏实现）：

• 频道输入：st.session_state.youtube_channels 初始为一个空输入框，点击 "Add Channel" 可追加任意多个频道 URL，每个非首项都可删除；
• 日期范围：起止日期经 strftime("%Y-%m-%d") 转为字符串存入 session state，最终作为 start_date / end_date 传给 BrightData；
• 开始分析：主按钮触发 start_analysis()，页面会依次显示 "Scraping videos…"、轮询状态、提取出的视频列表（每行 3 个视频的响应式网格，用 st.video 内嵌播放）、字幕处理进度，以及智能体分析阶段，最后渲染报告并附下载按钮。

四、数据采集层：BrightData Datasets v3 的三步调用

brightdata_scrapper.py 封装了 BrightData 异步数据平台（先触发任务、拿快照 ID、轮询进度、再取结果）的完整生命周期，共四个函数。

4.1 触发采集：trigger_scraping_channels

trigger_scraping_channels 是本项目实际使用的入口：

dataset_id = "gd_lk56epmy2i5g7lzu0k"
endpoint = (f"https://api.brightdata.com/datasets/v3/trigger"
            f"?dataset_id={dataset_id}&include_errors=true"
            f"&type=discover_new&discover_by=url")

要点：

• 固定 dataset_id：项目绑定了一个预配置的 BrightData 数据集 gd_lk56epmy2i5g7lzu0k，用于"给定频道 URL 发现新视频"。替换采集范围需要更换该 ID；
• 逐频道构造 payload：每个频道 URL 生成一条记录，字段包括 url、num_of_posts（app.py#L108 中固定传 10，即每频道最多 10 条）、start_date、end_date、order_by（固定传 "Latest"）、country（传空字符串）；
• 认证与请求：以 Authorization: Bearer {api_key} 头发起 JSON POST。源码选择用 subprocess 拼 curl 命令执行（brightdata_scrapper.py#L59-L75），而非 Python HTTP 库——从源码结构看这是一种不引入额外依赖的轻量做法，返回值为解析后的 JSON（触发成功时包含 snapshot_id）。

同文件还保留了 trigger_scraping_niche：按关键词（keyword）而非频道 URL 触发采集，payload 中 endpoint 由调用方传入。当前 app.py 未调用它，可视为"按关键词挖趋势"这一同类能力的预留扩展。

4.2 轮询进度：get_progress

def get_progress(api_key, snapshot_id):
    command = ["curl", "-H", f"Authorization: Bearer {api_key}",
               f"https://api.brightdata.com/datasets/v3/progress/{snapshot_id}"]

get_progress 按 snapshot_id 查询任务状态。app.py#L109-L118 中的消费方式是典型的状态机轮询：

status = get_progress(bright_data_api_key, channel_snapshot_id['snapshot_id'])
while status['status'] != "ready":
    status_container.info(f"Current status: {status['status']}")
    time.sleep(10)
    status = get_progress(bright_data_api_key, channel_snapshot_id['snapshot_id'])
    if status['status'] == "failed":
        status_container.error(f"Scraping failed: {status}")
        return

每 10 秒刷新一次；状态为 ready 进入下一阶段，failed 则终止并把错误透出到界面，其余中间状态（如排队、运行中）实时显示在 st.empty() 占位容器中。

4.3 拉取结果：get_output

get_output 请求 datasets/v3/snapshot/{snapshot_id}?format=json，返回内容按行分割后逐行 json.loads，即标准的 JSON Lines 解析。从 app.py#L137-L151 的消费方式看，返回对象形如 [ [video, video, ...] ]：channel_scrapped_output[0] 是视频列表，每条视频包含 url、shortcode（视频 ID）与 formatted_transcript 等字段，其中字幕条目含 text、start_time、end_time 三个键（app.py#L166-L170）。

五、字幕落盘：为 FileReadTool 准备"证据文件"

采集完成后，app.py#L153-L172 把每支视频的字幕写成独立文本文件：

youtube_video_id = channel_scrapped_output[0][i]['shortcode']
file = "transcripts/" + youtube_video_id + ".txt"
st.session_state.all_files.append(file)
with open(file, "w") as f:
    for j in range(len(channel_scrapped_output[0][i]['formatted_transcript'])):
        text = channel_scrapped_output[0][i]['formatted_transcript'][j]['text']
        start_time = channel_scrapped_output[0][i]['formatted_transcript'][j]['start_time']
        end_time = channel_scrapped_output[0][i]['formatted_transcript'][j]['end_time']
        f.write(f"({start_time:.2f}-{end_time:.2f}): {text}\n")

两个值得注意的设计：

1. 保留时间轴：每行字幕带 (start-end): 文本 前缀，智能体分析时可引用"某时间点说了什么"；
2. 文件路径驱动而非文本直传：所有路径被收集进 st.session_state.all_files，最终作为逗号分隔字符串 file_paths 传给 crew.kickoff()。这正是 CrewAI FileReadTool 的典型用法——分析 Agent 拿到的是路径列表，由工具按需读取文件内容（app.py#L15-L17 中 docs_tool = FileReadTool()，并只挂载给第一个分析 Agent），长字幕不必一次性塞进 prompt。

注意 transcripts/ 目录由代码直接写入而未先创建，首次运行前需自行 mkdir transcripts，否则写文件会抛 FileNotFoundError。

六、多智能体分析层：两个 Agent 与顺序执行流程

6.1 配置即代码：config.yaml

config.yaml 将 Agent 与 Task 的提示词全部外置，结构为顶层 agents 与 tasks 两个列表。

Agent 1 —— analysis_agent（YouTube Transcript Analyzer）

• goal：分析 {file_paths} 中的多个视频字幕，产出分节报告：① 关键话题；② 跨多条字幕的新兴趋势/模式；③ 说话者情绪与语气；④ 反复出现的关键词或短语；
• backstory：设定为"细致、强分析力的专家，擅长从复杂 YouTube 字幕中提取说话者意图、主题、趋势与情绪"；
• verbose: true 便于运行时观察推理过程。

Agent 2 —— response_synthesizer_agent（Response Synthesizer Agent）

• goal：把第一节分析浓缩为连贯、简洁的回应——总结各节关键发现、给出可执行建议、保持清晰与可读；
• backstory：擅长把深度分析翻译成简洁可行动语言、面向决策场景的沟通者。

Task 定义与 Agent 一一对应（agent 字段在 YAML 中显式声明）：

Task	description 摘要	expected_output 摘要
analysis_task	对 {file_paths} 的字幕做细粒度四节分析	含四节（主题/趋势/情绪/关键词）的详尽多节报告，各节带示例与上下文解释
response_task	将分节分析综合为简明摘要，突出可执行建议	简洁摘要：各节高层发现 + 行动建议 + 决策导向的清晰语言

{file_paths} 是占位符：Task 1 的 description 中引用它，由 kickoff(inputs={"file_paths": ...}) 在运行时注入真实文件列表（app.py#L186）。

6.2 Crew 的构建与执行

create_agents_and_tasks() 用 yaml.safe_load 读取配置并逐项组装：

crew = Crew(
    agents=[analysis_agent, response_synthesizer_agent],
    tasks=[analysis_task, response_task],
    process=Process.sequential,
    verbose=True
)

关键决策是 Process.sequential（顺序流程）：analysis_task 必须先完整产出四节报告，response_task 在其结果之上做综合——这是"先深挖、后收敛"的两阶段模式，保证最终摘要的信息来自细粒度中间产物，而不是让单一模型一次性硬做长文总结。注意一个源码细节：Agent 构造时 role/goal/backstory 直接从 config["agents"][i] 的对应键读取，而 verbose 参数硬编码为 True（未读取 YAML 中的 verbose: true 字段）；LLM 两个 Agent 共用同一个 load_llm() 实例。

执行入口在 start_analysis() 后半段：

st.session_state.crew = create_agents_and_tasks()
st.session_state.response = st.session_state.crew.kickoff(
    inputs={"file_paths": ", ".join(st.session_state.all_files)})

kickoff 的返回值（CrewOutput）存入 session state，主界面区（app.py#L247-L260）用 st.markdown(result) 渲染报告正文，并用 result.raw 提供 "Download Content" 按钮，下载文件名为 youtube_trend_analysis.md。

七、适用前提与注意事项

• Python ≥ 3.11：README 明确的前提；
• BrightData 账号与有效 dataset_id：采集依赖预配置的 gd_lk56epmy2i5g7lzu0k 数据集，API Key 与 dataset 必须匹配可用；
• LLM 二选一：默认走 OpenAI gpt-4o（需 OPENAI_API_KEY），Ollama 本地路线（llama3.2 @ http://localhost:11434）需手动切换 app.py#L24-L29 中的注释行；
• 每频道上限与排序：当前硬编码每频道 10 条视频、按 Latest 排序（app.py#L108），调整需在调用处修改参数；
• transcripts 目录：代码直接向 transcripts/ 写字幕文件，首次运行前请确认目录存在；
• 运行方式：仅介绍查看与运行方式，本项目源码位于仓库 Youtube-trend-analysis/ 目录下，streamlit run app.py 需在该目录内执行。

至此，从 README 的三步安装（API Key / Ollama / 依赖）与一条启动命令出发，我们沿着 app.py 的编排逻辑、brightdata_scrapper.py 的 API 封装与 config.yaml 的智能体提示词，完整还原了这条"BrightData 采集 → 字幕落盘 → CrewAI 顺序双智能体分析 → Streamlit 呈现下载"的 YouTube 趋势分析流水线，可作为多智能体 + 网页数据源组合应用的参考实现。