第8课：完整实战 —— 从零搭建智能研究平台
本课目标
前面 7 课学了所有零件，这节课把它们组装起来，搭一个完整的智能研究平台。
项目结构
my-research-agent/
├── .env ← API Key
├── pyproject.toml ← 依赖配置
├── research_agent/
│ ├── agent.py ← 主程序（入口）
│ ├── prompts/ ← 各角色的提示词
│ │ ├── lead_agent.txt
│ │ ├── researcher.txt
│ │ ├── data_analyst.txt
│ │ └── report_writer.txt
│ └── utils/
│ ├── subagent_tracker.py ← 子Agent追踪器
│ ├── message_handler.py ← 消息处理器
│ └── transcript.py ← 日志系统
├── files/ ← 运行时生成的文件
│ ├── research_notes/ ← 研究笔记
│ ├── charts/ ← 图表
│ ├── data/ ← 数据摘要
│ └── reports/ ← 最终报告
└── logs/ ← 会话日志
我们按从外到内的顺序来搭建。
第一步：写提示词（Prompts）
提示词是整个系统的灵魂。每个角色一个文件。
lead_agent.txt — 组长
你是一个研究项目的协调员。你的唯一工作是拆解任务并调度子Agent。
重要规则：
- 你只能使用 Task 工具来派活
- 你自己不搜索、不写文件、不分析
- 回复要简短，2-3句话就够
工作流程：
1. 收到用户的研究请求后，拆成 2-4 个具体子课题
2. 用 Task 工具同时派出多个 researcher，每个负责一个子课题
3. 等所有 researcher 完成后，派出 data-analyst 分析数据并生成图表
4. 等 data-analyst 完成后，派出 report-writer 生成最终 PDF 报告
5. 告诉用户报告路径
派活格式：
Task(subagent_type="researcher", description="简短描述", prompt="详细指令")
Task(subagent_type="data-analyst", description="简短描述", prompt="详细指令")
Task(subagent_type="report-writer", description="简短描述", prompt="详细指令")
注意：researcher 要同时派出（并行），data-analyst 和 report-writer 要顺序派出。
researcher.txt — 研究员
你是一个数据驱动的研究员。
核心要求：
- 必须使用 WebSearch 搜索 5-10 次
- 优先搜索数字和数据：市场规模、增长率、排名、百分比
- 不许用自己的知识，所有信息必须来自搜索结果
- 每条数据要标注来源
工作流程：
1. 围绕指定主题，构造 5-10 个搜索关键词
2. 用 WebSearch 逐一搜索
3. 从搜索结果中提取关键数据和统计数字
4. 整理成 markdown 格式的研究笔记
5. 用 Write 工具保存到 files/research_notes/ 目录
输出格式：
# [主题] 研究笔记
## 关键数据
- 市场规模：$XX billion (来源)
- 增长率：XX% CAGR (来源)
- 市场份额：XX% (来源)
## 主要发现
1. [发现1，带具体数字]
2. [发现2，带具体数字]
## 数据对比表
| 项目 | 数值 | 来源 |
|------|------|------|
## 信息来源
- [来源名]: URL
质量标准：至少包含 10 个具体数字。
data_analyst.txt — 数据分析师
你是一个数据分析师，负责将研究笔记转化为可视化图表。
工作流程：
1. 用 Glob 找到 files/research_notes/ 下的所有 .md 文件
2. 用 Read 读取每个文件
3. 提取所有数字数据（市场规模、增长率、排名等）
4. 根据数据类型选择合适的图表：
- 对比数据 → 柱状图
- 时间趋势 → 折线图
- 占比分布 → 饼图
- 排名 → 水平柱状图
5. 用 Bash 执行 Python (matplotlib) 生成 2-4 张图表
6. 保存图表到 files/charts/ 目录（PNG 格式）
7. 写一份数据摘要到 files/data/data_summary.md
图表生成模板（用 Bash 工具执行）：
python3 << 'EOF'
import matplotlib.pyplot as plt
import os
os.makedirs('files/charts', exist_ok=True)
# 设置中文字体
# macOS 用 'PingFang SC'，Linux 用 'Noto Sans CJC SC'，Windows 用 'SimHei'
# 如果不确定，可以先用英文标签避免乱码
import platform
if platform.system() == 'Darwin': # macOS
plt.rcParams['font.sans-serif'] = ['PingFang SC', 'Arial Unicode MS']
elif platform.system() == 'Linux':
plt.rcParams['font.sans-serif'] = ['Noto Sans CJK SC', 'WenQuanYi Micro Hei']
else: # Windows
plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei']
plt.rcParams['axes.unicode_minus'] = False # 解决负号显示为方块的问题
fig, ax = plt.subplots(figsize=(10, 6))
# ... 画图代码 ...
plt.savefig('files/charts/图表名.png', dpi=150, bbox_inches='tight')
plt.close()
print("已保存: files/charts/图表名.png")
EOF
注意：如果运行环境没装中文字体，最保险的做法是让 Claude 用英文写图表标签。
report_writer.txt — 报告撰写人
你是一个专业的报告撰写人。
工作流程：
1. 用 Glob 找到以下目录中的所有文件：
- files/research_notes/（研究笔记）
- files/data/（数据摘要）
- files/charts/（图表）
2. 用 Read 读取所有文本文件
3. 综合所有材料，生成一份 PDF 报告
4. 用 Bash 执行 Python (reportlab) 生成 PDF
5. 保存到 files/reports/
报告结构：
- 标题和日期
- 执行摘要（200字以内）
- 关键发现（3-5点，带数据）
- 详细分析（嵌入图表）
- 信息来源
要求：1-2 页，500-1000 字，专业排版。
你不搜索、不分析——只整合已有材料。
第二步：子Agent追踪器
这是研究平台的"监控中心"：
"""
research_agent/utils/subagent_tracker.py
追踪所有子Agent的活动
"""
import json
from datetime import datetime
from pathlib import Path
class SubagentTracker:
def __init__(self, log_dir: Path):
self.sessions = {} # {parent_tool_use_id: session_info}
self._current_parent_id = None
self._counters = {} # {"researcher": 2, "data-analyst": 1, ...}
self._log_file = log_dir / "tool_calls.jsonl"
def register_subagent_spawn(self, tool_use_id, subagent_type, description, prompt):
"""当组长派出一个子Agent时调用"""
# 生成唯一ID：RESEARCHER-1, RESEARCHER-2, DATA-ANALYST-1, ...
count = self._counters.get(subagent_type, 0) + 1
self._counters[subagent_type] = count
agent_id = f"{subagent_type.upper()}-{count}"
self.sessions[tool_use_id] = {
"id": agent_id,
"type": subagent_type,
"description": description,
"spawned_at": datetime.now().isoformat(),
"tool_calls": [],
}
return agent_id
def set_current_context(self, parent_tool_use_id):
"""设置当前正在工作的子Agent上下文"""
self._current_parent_id = parent_tool_use_id
async def pre_tool_use_hook(self, hook_input, tool_use_id, context):
"""Hook：工具调用前"""
tool_name = hook_input["tool_name"]
tool_input = hook_input["tool_input"]
# Task 工具由消息处理器单独处理
if tool_name == "Task":
return {}
# 找到当前子Agent
agent_id = "LEAD"
if self._current_parent_id and self._current_parent_id in self.sessions:
session = self.sessions[self._current_parent_id]
agent_id = session["id"]
session["tool_calls"].append({
"tool": tool_name,
"timestamp": datetime.now().isoformat(),
})
# 打印到控制台
print(f" [{agent_id}] → {tool_name}")
# 写入 JSONL 日志
log_entry = {
"event": "tool_call",
"agent_id": agent_id,
"tool_name": tool_name,
"timestamp": datetime.now().isoformat(),
}
with open(self._log_file, "a") as f:
f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")
return {}
async def post_tool_use_hook(self, hook_input, tool_use_id, context):
"""Hook：工具调用后"""
tool_name = hook_input["tool_name"]
# 可以在这里记录结果、统计耗时等
return {}
第三步：消息处理器
负责从消息流中识别子Agent的派出：
"""
research_agent/utils/message_handler.py
处理消息流，识别子Agent的创建
"""
from claude_agent_sdk import AssistantMessage, TextBlock, ToolUseBlock
def process_assistant_message(msg, tracker, print_fn):
"""处理一条 AssistantMessage"""
# 更新当前上下文（知道现在是哪个子Agent在工作）
parent_id = getattr(msg, 'parent_tool_use_id', None)
if parent_id:
tracker.set_current_context(parent_id)
if not isinstance(msg, AssistantMessage):
return
for block in msg.content:
# 文字内容：直接打印
if isinstance(block, TextBlock):
print_fn(block.text)
# 工具调用：检查是不是在派活
elif isinstance(block, ToolUseBlock):
if block.name == "Task":
# 组长在派子Agent！
agent_type = block.input.get("subagent_type", "unknown")
description = block.input.get("description", "")
prompt = block.input.get("prompt", "")
agent_id = tracker.register_subagent_spawn(
tool_use_id=block.id,
subagent_type=agent_type,
description=description,
prompt=prompt,
)
print_fn(f"\n🚀 派出 [{agent_id}]: {description}")
第四步：主程序
把所有零件组装起来：
"""
research_agent/agent.py
主程序 —— 智能研究平台入口
"""
import anyio
from pathlib import Path
from datetime import datetime
from dotenv import load_dotenv
from claude_agent_sdk import (
ClaudeSDKClient, ClaudeAgentOptions, AgentDefinition,
HookMatcher, AssistantMessage
)
# 加载自己写的模块
from utils.subagent_tracker import SubagentTracker
from utils.message_handler import process_assistant_message
load_dotenv()
def load_prompt(filename: str) -> str:
"""从 prompts/ 目录加载提示词"""
path = Path(__file__).parent / "prompts" / filename
return path.read_text(encoding="utf-8")
async def main():
# ===== 1. 创建目录结构 =====
project_root = Path(__file__).parent.parent.resolve()
session_name = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
log_dir = project_root / "logs" / session_name
log_dir.mkdir(parents=True, exist_ok=True)
# 预创建文件目录，子Agent会往这里写文件
for subdir in ["research_notes", "charts", "data", "reports"]:
(project_root / "files" / subdir).mkdir(parents=True, exist_ok=True)
print(f"📁 项目目录: {project_root}")
print(f"📁 日志目录: {log_dir}")
# ===== 2. 初始化追踪器 =====
tracker = SubagentTracker(log_dir=log_dir)
# ===== 3. 加载提示词 =====
lead_prompt = load_prompt("lead_agent.txt")
researcher_prompt = load_prompt("researcher.txt")
analyst_prompt = load_prompt("data_analyst.txt")
writer_prompt = load_prompt("report_writer.txt")
# ===== 4. 定义子Agent =====
agents = {
"researcher": AgentDefinition(
description="搜索互联网收集资料的研究员",
prompt=researcher_prompt,
tools=["WebSearch", "Write"],
model="haiku",
),
"data-analyst": AgentDefinition(
description="分析数据并生成图表的数据分析师",
prompt=analyst_prompt,
tools=["Glob", "Read", "Bash", "Write"],
model="haiku",
),
"report-writer": AgentDefinition(
description="整合材料生成PDF报告的撰写人",
prompt=writer_prompt,
tools=["Write", "Glob", "Read", "Bash"],
model="haiku",
),
}
# ===== 5. 设置 Hooks =====
hooks = {
"PreToolUse": [
HookMatcher(matcher=None, hooks=[tracker.pre_tool_use_hook]),
],
"PostToolUse": [
HookMatcher(matcher=None, hooks=[tracker.post_tool_use_hook]),
],
}
# ===== 6. 配置选项 =====
options = ClaudeAgentOptions(
system_prompt=lead_prompt,
allowed_tools=["Task"], # 组长只能派活
agents=agents,
hooks=hooks,
permission_mode="bypassPermissions",
model="sonnet", # 组长用好一点的模型
max_turns=20, # 留足够轮次
cwd=str(project_root), # 工作目录，确保文件写到项目里
)
# ===== 7. 启动聊天循环 =====
async with ClaudeSDKClient(options=options) as client:
print("\n=== 🔬 智能研究平台已启动 ===")
print("输入你的研究课题，输入 q 退出\n")
while True:
user_input = input("你: ").strip()
if user_input.lower() in ["q", "quit", "exit"]:
break
if not user_input:
continue
print(f"\n{'='*50}")
print(f"开始研究: {user_input}")
print(f"{'='*50}\n")
await client.query(user_input)
async for msg in client.receive_response():
process_assistant_message(
msg, tracker,
print_fn=lambda text: print(f" 💬 {text}")
)
print(f"\n{'='*50}")
print("研究完成！")
print(f"日志: {log_dir}")
print(f"{'='*50}\n")
anyio.run(main)
第五步：运行！
# 确保在项目根目录
cd my-research-agent
# 运行
uv run python research_agent/agent.py
完整的交互过程大概是这样的：
📁 项目目录: /Users/you/my-research-agent
📁 日志目录: /Users/you/my-research-agent/logs/session_20260228_143022
=== 🔬 智能研究平台已启动 ===
输入你的研究课题，输入 q 退出
你: 帮我研究一下全球电动汽车市场
==================================================
开始研究: 帮我研究一下全球电动汽车市场
==================================================
💬 我将把这个课题拆分为4个方向来研究。
🚀 派出 [RESEARCHER-1]: 全球电动汽车市场规模和增长趋势
[RESEARCHER-1] → WebSearch
[RESEARCHER-1] → WebSearch
[RESEARCHER-1] → WebSearch
[RESEARCHER-1] → WebSearch
[RESEARCHER-1] → WebSearch
[RESEARCHER-1] → Write
🚀 派出 [RESEARCHER-2]: 主要电动汽车制造商和市场份额
[RESEARCHER-2] → WebSearch
[RESEARCHER-2] → WebSearch
[RESEARCHER-2] → WebSearch
[RESEARCHER-2] → Write
🚀 派出 [RESEARCHER-3]: 电池技术发展和成本趋势
[RESEARCHER-3] → WebSearch
[RESEARCHER-3] → WebSearch
[RESEARCHER-3] → Write
🚀 派出 [DATA-ANALYST-1]: 分析数据并生成图表
[DATA-ANALYST-1] → Glob
[DATA-ANALYST-1] → Read
[DATA-ANALYST-1] → Read
[DATA-ANALYST-1] → Read
[DATA-ANALYST-1] → Bash
[DATA-ANALYST-1] → Bash
[DATA-ANALYST-1] → Write
🚀 派出 [REPORT-WRITER-1]: 生成PDF研究报告
[REPORT-WRITER-1] → Glob
[REPORT-WRITER-1] → Read
[REPORT-WRITER-1] → Read
[REPORT-WRITER-1] → Bash
[REPORT-WRITER-1] → Write
💬 研究完成。报告已保存到 files/reports/ev_market_report_20260228.pdf
==================================================
研究完成！
日志: logs/session_20260228_143022
==================================================
运行后的文件结构
my-research-agent/
├── files/
│ ├── research_notes/
│ │ ├── ev_market_size.md ← RESEARCHER-1 的笔记
│ │ ├── ev_manufacturers.md ← RESEARCHER-2 的笔记
│ │ └── battery_technology.md ← RESEARCHER-3 的笔记
│ ├── charts/
│ │ ├── market_share.png ← 市场份额饼图
│ │ └── sales_growth.png ← 销量增长折线图
│ ├── data/
│ │ └── data_summary.md ← 数据分析摘要
│ └── reports/
│ └── ev_market_report_20260228.pdf ← 最终报告！
└── logs/
└── session_20260228_143022/
├── transcript.txt ← 人类可读的完整记录
└── tool_calls.jsonl ← 机器可解析的工具调用日志
核心设计回顾
搭完了，我们回头看看这个系统用到了教程里的哪些知识：
| 课程 | 在平台里体现为 |
|---|---|
| 第3课 query() | 基础通信方式 |
| 第4课 ClaudeSDKClient | 主程序的交互循环 |
| 第5课 Tools | 每个角色的工具分配 |
| 第6课 子Agent | researcher / analyst / writer 定义 |
| 第7课 Hooks | SubagentTracker 的追踪和日志 |
这不是堆砌技术，每个零件都有它的位置。
下一课预告
平台跑起来了，但还有很多优化空间——日志怎么更详细？花了多少钱怎么追踪？出错了怎么处理？