手把手带你用1500行Python代码打造DeerFlow style全托管AI写作流水线!(文末有token福利，先到先得)

烟雨平生

发布于 2026-04-14 19:14:35

370

从信息搜集到草稿上传，10 步自动化，让 AI 替你完成 80% 的写作工作

导语

技术公众号作者都经历过这样的痛点：每天花 2-3 小时翻阅 Hacker News、GitHub Trending、RSS 订阅源寻找选题，再花 2 小时查阅资料、撰写文章，最后还要排版、做封面图、登录公众号后台上传草稿。一篇文章从选题到发布，至少半天时间。

更令人沮丧的是，投入大量时间写的文章，阅读量可能并不理想——因为选题靠"感觉"，缺乏数据支撑；内容靠"经验"，容易编造不存在的 API 或命令。

这篇文章分享一套完全落地的解决方案：用 Python + LLM 打造一条全自动的文章写作流水线。从信息搜集、选题评分、深度调研、AI 写作、事实校验、代码沙箱验证、HTML 排版、封面图生成，到最终上传微信草稿箱——全程无需人工干预，只需要一条命令：

uv run python -m src.main

一、整体架构：10 步流水线

整条流水线的设计思路是"像工厂流水线一样处理每一篇文章"。每个步骤职责单一、可独立测试、失败不阻塞后续流程。

信息搜集 → LLM筛选评分 → 智能选题 → 深度调研 → AI写作
    ↓
上传草稿 ← 生成封面 ← HTML排版 ← 事实+代码校验 ← 提取标题摘要

对应到代码层面，核心编排逻辑只有不到 100 行：

def run_pipeline(config_path=None, skip_gather=False, topic_title=None):
    config = load_config(config_path)
    db = TopicDB()
    llm = LLMClient(config["llm"])

    # Step 1: 信息搜集（HN + GitHub + RSS 三源并行）
    all_items = []
    if sources.get("hackernews", {}).get("enabled"):
        all_items.extend(fetch_top_stories(top_n=10))
    if sources.get("github_trending", {}).get("enabled"):
        all_items.extend(fetch_trending(languages="python,typescript,go,rust"))
    if sources.get("rss", {}).get("enabled"):
        all_items.extend(fetch_rss(feeds))

    # Step 2: LLM 筛选评分
    scored = filter_and_score(all_items, llm)

    # Step 3: 智能选题
    topic = selector.select_topic()

    # Step 3.5: 深度调研
    research_summary = research_topic(topic, llm)

    # Step 4: AI 写作
    article_md = generate_article(topic, llm, author, research_summary)

    # Step 5: 提取标题摘要
    title = extract_title(article_md, llm)
    digest = extract_summary(article_md, llm)

    # Step 5.5: 事实校验 + 代码沙箱验证
    is_valid, feedback = verify_article(article_md, research_summary, llm)

    # Step 6: 转换微信 HTML
    article_html, _ = format_article(article_md, title)

    # Step 7: AI 生成封面图
    cover_path = img_gen.run(title, insight, output_dir)

    # Step 8: 上传草稿箱
    media_id = publisher.publish_article(title, article_html, cover_path, author, digest)

每个步骤用一个模块封装，模块之间通过字典（topic、article_md、research_summary）传递数据，没有任何隐式耦合。

二、三源并行信息搜集

第一步是信息搜集，我们同时接入三个技术圈最主流的信息源：

Hacker News —— 用 Firebase API 并发拉取 Top N 帖子。通过 ThreadPoolExecutor 并发请求，10 条数据不到 2 秒：

def fetch_top_stories(top_n=10) -> list:
    resp = requests.get(
        "https://hacker-news.firebaseio.com/v0/topstories.json", timeout=15
    )
    ids = resp.json()[:top_n]

    with ThreadPoolExecutor(max_workers=5) as executor:
        future_to_id = {executor.submit(fetch_item, item_id): item_id for item_id in ids}
        for future in as_completed(future_to_id):
            item = future.result()
            if item:
                items.append(item)
    return items

GitHub Trending —— 爬取 GitHub 趋势页面，支持按语言筛选。通过 BeautifulSoup 解析 article.Box-row 提取仓库名、描述和 Star 数。

RSS 订阅源 —— 用 feedparser 解析任意 RSS Feed，支持 InfoQ、36氪等中文技术媒体。

三个数据源的结果统一为标准格式 {"title", "url", "source", "summary", "score"}，汇聚后进入下一步筛选。

三、LLM 筛选评分与智能选题

搜集到的原始信息可能有 30-50 条，不可能每条都写。我们用 LLM 做两轮筛选。

第一轮：批量评分

把所有条目打包发给 LLM，让它在一次调用中完成所有评分。每条返回一个 0-10 的综合评分，评估维度包括时效性、实用性、深度和受众匹配度：

SYSTEM_PROMPT = """你是一位技术内容筛选专家。对每条技术信息进行评估：
1. 用一句话总结（中文）
2. 分类：AI/架构/前端/后端/DevOps/安全/开源/其他
3. 评分：0-10（综合时效性、实用性、深度、受众匹配度）
4. 核心洞见：一句话说明这件事为什么重要

返回JSON数组，每个元素格式：
{"title": "原标题", "summary": "中文总结", "category": "分类",
 "score": 分数, "insight": "核心洞见"}"""

只有评分 >= 6.0 的条目才会进入主题池（SQLite 数据库存储），避免浪费 Token。

第二轮：选题决策

从主题池中取出 Top 10 候选，让 LLM 扮演"选题编辑"角色，综合考虑时效性、实用性、深度、受众匹配、差异化五个维度，选出今天最适合发布的 1 个主题：

SELECTOR_PROMPT = """你是公众号的选题编辑。从候选主题中选出最适合今天发布的1个主题。

评估维度（各权重20%）：
1. 时效性 - 是否当下热点
2. 实用性 - 读者能否直接用上
3. 深度 - 能否提炼独特洞见
4. 受众匹配 - 3-10年经验的技术人是否关心
5. 差异化 - 和其他公众号是否有区别
"""

选题结果附带一句"核心洞见"（insight），这会成为后续写作和封面图生成的核心线索。

四、深度调研：借鉴 DeerFlow 的 Research 思路

这是整条流水线最关键的步骤，也是区别于"直接让 LLM 写文章"的核心差异。

为什么需要深度调研？ 直接让 LLM 写技术文章，它大概率会编造不存在的安装命令、伪造 API 名称、虚构配置示例。DeerFlow 项目的经验告诉我们：必须先用真实信息"喂饱" LLM，再让它基于这些信息写作。

调研流程分四步：

Step 1：LLM 生成搜索关键词

不是简单地把标题丢给搜索引擎，而是让 LLM 生成 3-5 个精准的搜索关键词：

RESEARCH_KEYWORD_PROMPT = """给定以下技术选题，请生成 3-5 个适合搜索引擎的查询关键词。

要求：
- 英文关键词为主（技术内容英文资源更丰富）
- 包含项目名/技术名的精确匹配
- 包含 "tutorial"、"example"、"github" 等限定词
- 返回 JSON 数组格式：["keyword1", "keyword2", ...]"""

Step 2：多关键词搜索 + 页面抓取

对每个关键词调用 DuckDuckGo Lite 搜索，取 Top 5 结果，然后逐个抓取页面完整正文：

for kw in keywords[:3]:
    results = _duckduckgo_search(kw, max_results=5)
    for r in results:
        content = _fetch_page_content(r["url"])
        if content and len(content) > 200:
            all_content.append(f"【来源: {r['title']}】\n{content}")

页面抓取会智能提取 <article> 或 <main> 区域的内容，去掉导航、页脚等噪音，并限制最大 5000 字符避免 Token 爆炸。

Step 3：LLM 整理结构化报告

把所有抓取到的原始资料（可能上万字）交给 LLM，整理成包含项目简介、核心特性、安装命令、代码示例、架构说明的结构化报告。

关键原则：只使用资料中出现的信息，缺失的部分明确标注"未在调研资料中找到"。

Step 4：降级容错

如果 DuckDuckGo 搜索失败，自动降级到 Wikipedia API。如果所有搜索都失败，返回空字符串，不影响后续流程——只是写作时没有调研资料支撑。

五、AI 写作：DeerFlow 风格的 Prompt 工程

有了调研报告，就可以让 LLM 写文章了。写作 Prompt 的设计借鉴了 DeerFlow 的核心理念——真实性优先、实用性强、深度分析：

WRITER_SYSTEM = """核心理念（DeerFlow 风格）：
- **真实性优先**：所有技术细节必须基于调研资料，严禁编造
- **实用性强**：每篇文章都要给读者带来可操作的价值
- **深度分析**：不仅讲"是什么"，更要讲"为什么"和"怎么用"

写作要求：
- 目标读者：3-10年经验的技术人
- 字数：1500-2500字
- 结构：标题 → 导语 → 正文 → 实践建议 → 一句话总结"""

Prompt 中强制要求 LLM 遵守六条写作原则：

基于真实资料写作：所有技术细节必须来自调研资料
引用真实代码：直接引用调研资料中的代码片段，不修改不"优化"
诚实对待未知：信息缺失时明确说明，不猜测
代码要有上下文：标注用途、依赖和预期输出
避免过度承诺：客观描述效果，不夸大
实用导向：让读者看完知道"怎么用"

六、事实校验 + 代码沙箱：双重质量关卡

文章生成后，不是直接发布，而是经过两道校验关卡。

第一关：LLM 事实校验

把调研资料和文章同时交给 LLM，让它扮演 Reviewer 角色，逐条核对：

VERIFY_FACTS_PROMPT = """请对比「调研资料」和「文章内容」，检查文章是否存在：
1. 虚构的安装命令
2. 虚构的 API 或方法名
3. 与调研资料矛盾的事实陈述
4. 伪造的配置示例

返回 JSON：{"is_valid": true/false, "issues": [...], "suggestion": "..."}"""

第二关：代码沙箱验证

这是最有特色的一步。从文章中提取所有 Python 代码块，在真实的沙箱环境中逐个执行：

def sandbox_execute_python(code: str, timeout: int = 10) -> dict:
    # 安全检查：AST 分析，禁止文件操作、系统调用
    is_safe, reason = _is_safe_code(code)
    if not is_safe:
        return {"success": False, "error": f"Code safety check failed: {reason}"}

    # 独立临时目录中执行
    with tempfile.TemporaryDirectory(prefix="wechatsandbox_") as tmpdir:
        result = subprocess.run(
            ["python3", "-c", code],
            capture_output=True, text=True, timeout=timeout, cwd=tmpdir
        )

安全机制包括：

AST 级别的安全检查：通过 ast.parse 分析代码，白名单机制只允许标准库中的安全模块（math、json、re 等），禁止 open、exec、eval、__import__ 等危险调用
独立临时目录：每个代码块在独立的临时目录中执行，避免相互干扰
超时控制：10 秒超时，防止死循环
智能跳过：需要外部框架（Flask、LangChain 等）的代码自动跳过，不算失败

如果代码执行失败，文章会被标记为"需人工复核"，但不阻止发布——因为最终决策权在人。

七、微信 HTML 排版：全内联样式

微信公众号会剥离所有 <style> 标签和外部 CSS，所以所有样式必须内联。我们实现了一套完整的 Markdown 到微信 HTML 的转换器，包含：

橙色科技主题：用 #FF6A00 作为主色调，深色代码块背景 #282c34
语法高亮：基于 Token 的代码着色，支持 Python 和 JavaScript 关键字
响应式排版：最大宽度 578px，适配手机阅读

ORANGE = "#FF6A00"
CODE_BLOCK_BG = "#282c34"

INLINE_STYLES = {
    "h1": f"border-left: 4px solid {ORANGE}; padding-left: 12px; font-size: 24px;",
    "pre": f"background-color: {CODE_BLOCK_BG}; border-radius: 6px; padding: 16px;",
    "strong": f"color: {ORANGE}; font-weight: bold;",
    "blockquote": f"border-left: 3px solid {ORANGE}; background-color: #fff8f0;",
}

代码块中的换行用 <br> 标签替代（微信会忽略 white-space: pre），每个语法元素用带颜色的 <span> 包裹，实现了近似 IDE 的代码高亮效果。

八、AI 封面图 + 自动上传草稿

封面图生成调用智谱 CogView-3-Flash（免费），根据标题和核心洞见生成封面。生成后会自动裁剪为微信推荐的 900×383 比例，并去掉底部可能的水印区域。

草稿上传封装了完整的微信公众号 API 调用链：

获取 access_token → 上传封面图为永久素材 → 创建草稿

关键细节：

Token 自动缓存到本地文件，过期前 5 分钟自动刷新
标题限制 64 字节 UTF-8，摘要限制 120 字节 UTF-8，超出自动截断
请求体用 json.dumps(ensure_ascii=False).encode("utf-8") 编码，避免中文乱码
所有 API 调用都通过 tenacity 实现了指数退避重试（最多 3 次）

九、配置设计：环境变量隔离敏感信息

配置文件采用 ${ENV_VAR} 占位符格式，运行时自动解析为环境变量值：

wechat:
  app_id: "${WECHAT_APP_ID}"
  app_secret: "${WECHAT_APP_SECRET}"

llm:
  api_key: "${ZHIPU_API_KEY}"
  model: "glm-5-turbo"
  base_url: "https://open.bigmodel.cn/api/paas/v4"

解析逻辑递归遍历整个配置树，自动替换所有 ${...} 占位符：

def _resolve_env(value: str) -> str:
    if isinstance(value, str) and value.startswith("${") and value.endswith("}"):
        env_key = value[2:-1]
        env_value = os.environ.get(env_key)
        if env_value is None:
            raise ValueError(f"Required environment variable '{env_key}' is not set")
        return env_value
    return value

配置加载后还会做完整的结构校验——检查必填字段是否存在、文章字数范围是否合理等。