内容生产自动化工作流

# 调研信息汇总的自动化脚本示例
import requests
from openai import OpenAI
 
client = OpenAI()
 
def gather_research(topic: str) -> str:
    # 从多个 API 获取信息
    sources = []
 
    # 1. 新闻 API
    news_resp = requests.get(
        "https://newsapi.org/v2/everything",
        params={"q": topic, "sortBy": "relevancy", "apiKey": NEWS_API_KEY}
    )
    for article in news_resp.json()["articles"][:5]:
        sources.append(f"[新闻] {article['title']}: {article['description']}")
 
    # 2. 学术论文 API
    papers_resp = requests.get(
        "https://api.semanticscholar.org/graph/v1/paper/search",
        params={"query": topic, "limit": 3, "fields": "title,abstract"}
    )
    for paper in papers_resp.json()["data"]:
        sources.append(f"[论文] {paper['title']}: {paper['abstract']}")
 
    # 3. 用 LLM 汇总
    summary_prompt = f"以下是关于「{topic}」的调研素材，请提取核心论点、关键数据和值得深入的方向：\n\n"
    summary_prompt += "\n\n".join(sources)
 
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": summary_prompt}]
    )
    return response.choices[0].message.content

# 写作 prompt 模板配置
writing_prompt:
  role: "你是一位面向海外 SaaS 决策者的技术内容编辑"
  tone: "专业但不学术，用具体案例说明抽象概念，避免营销话术"
  structure:
    - "开头用具体场景引入，不超过 100 字"
    - "正文分 3-5 个小节，每节一个小标题"
    - "每节包含至少一个数据点或案例"
    - "结尾给出可执行的行动建议"
  constraints:
    - "不使用 'revolutionary'、'game-changing' 等夸张词汇"
    - "技术术语保留英文，首次出现时给出中文解释"
    - "全文 2000-3000 词"
  references: "{research_output}"

# n8n 工作流配置示例：内容发布自动化
nodes:
  - name: "审核通过触发"
    type: webhook
    config:
      path: "/content-approved"
 
  - name: "获取内容数据"
    type: httpRequest
    config:
      url: "https://cms.internal/api/content/{{ $json.contentId }}"
      method: GET
 
  - name: "发布到 WordPress"
    type: wordpress
    config:
      operation: createPost
      title: "{{ $json.title }}"
      content: "{{ $json.body }}"
      status: publish
 
  - name: "同步到 Twitter"
    type: twitter
    config:
      operation: createTweet
      text: "{{ $json.socialSummary }}"
 
  - name: "发送团队通知"
    type: slack
    config:
      channel: "#content-publish"
      text: "新内容已发布：{{ $json.title }}"

# 串行工作流编排示例
async def content_pipeline(topic: str):
    # 步骤 1：调研
    research = await gather_research(topic)
 
    # 步骤 2：生成大纲
    outline = await generate_outline(research)
 
    # 步骤 3：写作初稿
    draft = await write_draft(outline, research)
 
    # 步骤 4：自动预检
    precheck_result = await auto_precheck(draft)
    if not precheck_result.passed:
        # 预检不通过，自动修改或标记问题
        draft = await fix_issues(draft, precheck_result.issues)
 
    # 步骤 5：人工审核（等待）
    review_result = await wait_for_human_review(draft)
    if not review_result.approved:
        return {"status": "rejected", "feedback": review_result.feedback}
 
    # 步骤 6：发布
    publish_result = await publish_to_channels(review_result.final_version)
 
    # 步骤 7：入库
    await index_to_knowledge_base(review_result.final_version)
 
    return {"status": "published", "urls": publish_result.urls}

import asyncio
 
async def parallel_translate(content: str, target_langs: list[str]):
    tasks = [
        translate_content(content, lang)
        for lang in target_langs
    ]
    results = await asyncio.gather(*tasks)
    return {lang: result for lang, result in zip(target_langs, results)}
 
# 同时翻译三种语言
translations = await parallel_translate(
    final_draft,
    ["ja", "es", "fr"]
)

# 条件分支逻辑
content_review:
  - condition: "content_type == 'technical_guide'"
    actions:
      - "技术编辑审核准确性"
      - "法务审核是否涉及合规声明"
      - "SEO 优化检查"
 
  - condition: "content_type == 'blog_post'"
    actions:
      - "内容编辑审核质量和调性"
      - "SEO 优化检查"
 
  - condition: "content_type == 'social_media'"
    actions:
      - "内容编辑快速审核"
      - "品牌合规自动检查"
 
  - condition: "word_count > 5000"
    actions:
      - "增加资深编辑二轮审核"

# 人工介入点配置
review_gates = {
    "topic_approval": {
        "required": True,
        "reviewer": "content_lead",
        "sla": "24h"
    },
    "draft_review": {
        "required": True,
        "reviewer": "editor",
        "sla": "48h"
    },
    "translation_qa": {
        "required": False,
        "sampling_rate": 0.3,  # 抽检 30%
        "reviewer": "native_speaker",
        "sla": "12h"
    },
    "publish_approval": {
        "required": True,
        "reviewer": "content_lead",
        "sla": "4h"
    }
}

def auto_precheck(content: dict) -> dict:
    issues = []
 
    # 1. 基础检查
    if len(content["body"]) < MIN_WORD_COUNT:
        issues.append({"type": "warning", "msg": f"字数不足 {MIN_WORD_COUNT}"})
 
    # 2. 语法和拼写检查
    grammar_issues = check_grammar(content["body"])
    issues.extend(grammar_issues)
 
    # 3. 敏感词扫描
    sensitive_words = scan_sensitive_words(content["body"], SENSITIVE_WORD_LIST)
    if sensitive_words:
        issues.append({"type": "blocker", "msg": f"包含敏感词：{sensitive_words}"})
 
    # 4. 重复率检测
    similarity = check_similarity(content["body"], existing_content_db)
    if similarity > 0.7:
        issues.append({"type": "warning", "msg": f"与现有内容相似度过高：{similarity:.0%}"})
 
    # 5. 链接有效性检查
    broken_links = check_links(content["body"])
    if broken_links:
        issues.append({"type": "blocker", "msg": f"存在失效链接：{broken_links}"})
 
    # 6. 品牌术语一致性
    term_issues = check_brand_terms(content["body"], BRAND_TERMS)
    issues.extend(term_issues)
 
    return {
        "passed": not any(i["type"] == "blocker" for i in issues),
        "issues": issues
    }

# 反馈数据收集与分析
def analyze_content_performance(content_id: str) -> dict:
    metrics = {
        "views": get_page_views(content_id),
        "avg_read_time": get_avg_read_time(content_id),
        "bounce_rate": get_bounce_rate(content_id),
        "social_shares": get_social_shares(content_id),
        "conversion_rate": get_conversion_rate(content_id),
        "seo_ranking": get_keyword_ranking(content_id),
    }
 
    # 与同类内容对比
    benchmark = get_category_benchmark(content_id)
    performance = {}
    for metric, value in metrics.items():
        bench_value = benchmark[metric]
        performance[metric] = {
            "value": value,
            "benchmark": bench_value,
            "delta": (value - bench_value) / bench_value if bench_value else 0
        }
 
    return performance

translation_prompt = """将以下内容翻译为{target_lang}。
 
规则：
1. 品牌术语使用术语表：{glossary}
2. 保持原文的段落结构和标题层级
3. 代码示例中的注释也需翻译
4. 数字、日期格式适配目标语言的地区习惯
5. 不要翻译产品名称（如 Analytics Pro、DataSync）
6. 语气保持专业但不学术，适合技术决策者阅读
 
原文：
{content}
"""