把重复研究工作流交给 Agent：从 Claude Cowork 文章提炼出的可落地教程

适合对象：需要定期做市场扫描、技术周报、文献初筛、投资组合简报、竞品追踪的人。

核心目标：先把“重复收集与整理”自动化，不把“最终判断”交给 Agent。

来源与边界

一句话原则

不要一开始就让 Agent “自由研究一个主题”。先把任务缩小成：固定问题、固定来源、固定字段、固定报告格式、固定验收标准。

最小可行架构

一个可靠的研究 Agent 工作流可以拆成 5 层：

  1. 任务契约：本周要回答什么问题，哪些数据必须出现，哪些内容禁止越界。
  2. 来源清单：零售商、论文库、行情 API、内部文档、新闻源等。
  3. 采集层：搜索、连接器、MCP、API、爬虫或人工导入的 CSV/JSON。
  4. 整理层：去重、分组、摘要、风险标记、报告生成。
  5. 审核层：人工确认来源、异常、趋势解释和最终决策。

原文中的 Claude Cowork 主要负责第 3～4 层；真正的决策仍然在第 5 层。

三个适合先落地的场景

场景 1：每周硬件价格快照

原文事实：作者让 Cowork 每周并行搜索 10 个来源，从零售商、价格追踪器、分析师报告和媒体中收集硬件价格趋势，生成固定结构的 HTML 报告。

可落地改造：

场景 2：学术文献初筛

原文事实：作者使用 Apify Academic Research MCP Server 统一访问 arXiv、Google Scholar、PubMed 等来源，但 JSTOR 等付费墙数据库仍依赖机构访问权限。

可落地改造：

场景 3：个人或团队关注列表简报

原文事实：作者用 Financial Modeling Prep 连接器追踪自己持有的股票，只收集当前价格、估值、分析师共识、移动平均线和目标价，让信息处理从一小时缩短到几分钟。

可落地改造：

Mock-first 教学项目

先不用 Claude Cowork、MCP 或任何付费 API。下面用一个本地 JSON 文件模拟“采集结果”，验证报告流水线是否合理。

目录结构

weekly-research-agent/
  sources.json
  empty_sources.json
  research_workflow.py
  out_ok/
  out_empty/

第 1 步：准备样本输入

文件：sources.json

[
  {
    "category": "hardware_price",
    "source": "Retailer A",
    "title": "RTX 4070 street price",
    "value": "$549, down 3% week over week",
    "url": "https://example.com/retailer-a/rtx-4070"
  },
  {
    "category": "hardware_price",
    "source": "Analyst Note",
    "title": "GPU inventory trend",
    "value": "Channel inventory normalized",
    "url": "https://example.com/analyst/gpu-inventory"
  },
  {
    "category": "literature_review",
    "source": "arXiv",
    "title": "Agentic workflows for research assistants",
    "value": "New benchmark emphasizes source traceability",
    "url": "https://example.com/arxiv/agentic-workflows"
  },
  {
    "category": "portfolio_brief",
    "source": "Financial API",
    "title": "Example holding valuation",
    "value": "P/E within 5-year range; analyst consensus unchanged",
    "url": "https://example.com/finance/example-holding"
  }
]

文件：empty_sources.json

[]

第 2 步：实现报告生成脚本

文件：research_workflow.py

from __future__ import annotations

import argparse
import html
import json
from dataclasses import dataclass, asdict
from pathlib import Path


@dataclass(frozen=True)
class SourceItem:
    category: str
    source: str
    title: str
    value: str
    url: str


@dataclass(frozen=True)
class ReportSection:
    category: str
    findings: list[SourceItem]
    takeaway: str


def load_sources(path: Path) -> list[SourceItem]:
    raw_items = json.loads(path.read_text(encoding="utf-8"))
    return [SourceItem(**item) for item in raw_items]


def group_sources(items: list[SourceItem]) -> list[ReportSection]:
    grouped: dict[str, list[SourceItem]] = {}
    for item in items:
        grouped.setdefault(item.category, []).append(item)

    sections: list[ReportSection] = []
    for category, findings in grouped.items():
        takeaway = f"{category}: collected {len(findings)} checked source(s); review trend changes before action."
        sections.append(ReportSection(category=category, findings=findings, takeaway=takeaway))
    return sections


def render_html(sections: list[ReportSection], output_path: Path) -> None:
    body_parts: list[str] = ["<h1>Weekly Research Brief</h1>"]
    if not sections:
        body_parts.append("<p>No findings. Check source permissions, query terms, or connector limits.</p>")
    for section in sections:
        body_parts.append(f"<h2>{html.escape(section.category)}</h2>")
        body_parts.append(f"<p><strong>Takeaway:</strong> {html.escape(section.takeaway)}</p>")
        body_parts.append("<ul>")
        for item in section.findings:
            link = html.escape(item.url, quote=True)
            title = html.escape(item.title)
            source = html.escape(item.source)
            value = html.escape(item.value)
            body_parts.append(f'<li><a href="{link}">{title}</a> — {source}: {value}</li>')
        body_parts.append("</ul>")

    document = "<!doctype html><meta charset='utf-8'><title>Weekly Research Brief</title>" + "\n".join(body_parts)
    output_path.write_text(document, encoding="utf-8")


def write_trace(sections: list[ReportSection], trace_path: Path) -> None:
    trace = {
        "section_count": len(sections),
        "finding_count": sum(len(section.findings) for section in sections),
        "sections": [asdict(section) for section in sections],
    }
    trace_path.write_text(json.dumps(trace, ensure_ascii=False, indent=2), encoding="utf-8")


def run(source_path: Path, output_dir: Path) -> int:
    output_dir.mkdir(parents=True, exist_ok=True)
    items = load_sources(source_path)
    sections = group_sources(items)
    render_html(sections, output_dir / "weekly_report.html")
    write_trace(sections, output_dir / "trace.json")
    print(f"report={output_dir / 'weekly_report.html'}")
    print(f"sections={len(sections)} findings={sum(len(section.findings) for section in sections)}")
    if not sections:
        print("warning=no_findings")
        return 2
    return 0


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Build a mock weekly research report.")
    parser.add_argument("--sources", required=True, type=Path)
    parser.add_argument("--output-dir", required=True, type=Path)
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    raise SystemExit(run(args.sources, args.output_dir))


if __name__ == "__main__":
    main()

第 3 步：运行正常样本

python3 research_workflow.py --sources sources.json --output-dir out_ok

预期输出：

report=out_ok/weekly_report.html
sections=3 findings=4

验收标准：

第 4 步：运行空结果样本

python3 research_workflow.py --sources empty_sources.json --output-dir out_empty

预期输出：

report=out_empty/weekly_report.html
sections=0 findings=0
warning=no_findings

脚本会返回退出码 2。这不是崩溃，而是有意设计的“需要人工检查”信号：可能是查询词太窄、权限不足、连接器失败、来源限流或当天确实没有结果。

如何替换成真实工作流

替换点 1：把 JSON 输入换成 API / MCP / 搜索结果

先保持 SourceItem 字段不变：

category, source, title, value, url

然后把采集层替换成：

不要同时改采集、整理、报告三层。一次只替换一层，方便定位问题。

替换点 2：加入任务契约

每个周期任务都应该有一段固定 brief，例如：

任务：生成每周硬件价格快照。
范围：只追踪 GPU、CPU、SSD 三类。
来源：指定零售商、价格追踪站、分析师报告，不使用论坛传言。
字段：当前价格、上周价格、涨跌幅、来源链接、异常说明。
输出：HTML 报告 + JSON trace。
禁止：不要给采购结论；只标记“需要人工审核”的异常。
验收：每条价格必须有来源链接；无结果时必须输出 no_findings。

替换点 3：加入人工审核字段

建议在最终报告中加入三个审核字段：

needs_review: true | false
risk_reason: 来源不一致 / 数据缺失 / 波动异常 / 权限不足
suggested_next_step: 人工核查链接 / 延后判断 / 补充来源 / 忽略

这样 Agent 不需要直接做决策，只需要把“哪里值得看”标出来。

一周落地路线

Day 1：选一个重复研究任务

只选一个，不要同时做三类。优先选择：

Day 2：写任务契约和字段 schema

先写字段，不写 Agent 提示词。字段稳定后，提示词和采集方式才有边界。

Day 3：用手工 JSON 跑通报告生成

使用本文的 Mock 脚手架，确认报告结构、空结果处理和 trace 文件都符合预期。

Day 4：接入一个真实来源

只接一个来源，例如一个 RSS、一个 API、一个导出的 CSV。不要一开始就并行接 10 个来源。

Day 5：加异常检测

最小规则即可：

Day 6：定时运行但不自动决策

可以让系统定时生成报告，但不要让它自动发采购建议、投资建议或生产变更建议。

Day 7：复盘并决定是否扩源

看三个指标：

安全与失败处理清单

可复制的 Agent brief 模板

你是一个研究整理 Agent，不是最终决策者。

目标：{每周要生成的报告名称}
范围：{追踪对象、时间范围、来源范围}
固定字段：{必须输出的字段列表}
来源要求：每条事实必须保留 URL 或内部文档 ID。
输出格式：HTML 报告 + JSON trace。
失败处理：如果来源不可访问、结果为空、字段缺失，必须标记 needs_review，不允许补写猜测值。
禁止事项：不要给最终采购/投资/生产决策；只输出事实、变化、异常和待人工确认项。
验收标准：{字段完整率、来源链接、异常标记、空结果行为}

什么时候不适合做成 Agent

本文验证记录

本教程中的 Mock 脚手架已在本地用 Python 3 标准库执行验证：

正常样本：exit_code=0, sections=3, findings=4, html_exists=True
空结果样本：exit_code=2, sections=0, findings=0, warning=no_findings

验证级别：

最后建议

把 Agent 当成“研究流水线的整理员”，不要当成“研究结论的负责人”。先让它稳定地产生可审计报告，再逐步扩大来源和自动化程度。