运营自动化里的 AI 工具：表格、邮件与工作流

# n8n 工作流核心逻辑（简化）
nodes:
  - name: "定时触发"
    type: schedule
    params: { cron: "0 9 * * 1" }  # 每周一上午 9 点
 
  - name: "拉取 Shopify 订单"
    type: httpRequest
    params:
      url: "https://{{shopify_store}}/admin/api/2026-01/orders.json"
      authentication: preDefinedCredential
      options: { paginate: true }
 
  - name: "拉取亚马逊订单"
    type: httpRequest
    params:
      url: "https://sellingpartnerapi.amazon.com/orders/v0/orders"
      # 签名逻辑在 credential 里处理
 
  - name: "AI 标准化字段"
    type: openAi
    params:
      model: gpt-4o-mini
      prompt: |
        将以下订单数据统一为标准格式。
        标准字段：order_id, platform, customer_email, amount, currency,
                 status, shipping_status, refund_status, created_at
        状态枚举映射：
        - 已发货/shipped/delivered → fulfilled
        - 待处理/pending/unshipped → pending
        - 已取消/cancelled → cancelled
        - 已退款/refunded/partial_refund → refunded
        输出 JSON 数组，严格按标准字段，缺失字段填 null。
 
  - name: "去重与合并"
    type: code
    params:
      javascript: |
        const allOrders = [...shopify, ...amazon, ...site]
        const deduped = new Map()
        for (const order of allOrders) {
          const existing = deduped.get(order.order_id)
          if (!existing || new Date(order.updated_at) > new Date(existing.updated_at)) {
            deduped.set(order.order_id, order)
          }
        }
        return [...deduped.values()]
 
  - name: "写入飞书多维表格（追加模式）"
    type: httpRequest
    params:
      url: "https://open.feishu.cn/open-apis/bitable/v1/apps/{{app_token}}/tables/{{table_id}}/records/batch_create"
      method: POST
      # 注意：是 batch_create 而非 batch_update

# 邮件处理工作流核心逻辑
from pydantic import BaseModel, Field
from enum import Enum
 
class EmailCategory(str, Enum):
    FAQ = "faq"                    # 常见问题
    FEATURE_REQUEST = "feature"    # 功能请求
    COMPLAINT = "complaint"        # 投诉
    TECHNICAL = "technical"        # 技术问题
    BILLING = "billing"            # 账单问题
    UNKNOWN = "unknown"            # 无法分类
 
class EmailClassification(BaseModel):
    category: EmailCategory
    confidence: float = Field(ge=0, le=1)
    summary: str
    suggested_reply: str | None = None
    needs_human: bool
 
# 分类 + 草稿生成 Prompt
CLASSIFY_PROMPT = """
你是客服邮件分类助手。根据邮件内容完成以下任务：
 
1. 分类到以下类别之一：{categories}
2. 给出置信度（0-1）
3. 用一句话总结邮件内容
4. 如果是 FAQ 或 BILLING 类，生成回复草稿（语气专业友好，不超过 150 字）
5. 判断是否需要人工处理（COMPLAINT、低置信度、涉及金额争议 → true）
 
邮件内容：
---
发件人：{{sender}}
主题：{{subject}}
正文：{{body}}
---
 
输出 JSON，严格匹配 schema。
"""
 
def process_email(sender: str, subject: str, body: str):
    # Step 1: AI 分类
    result = llm.classify(
        prompt=CLASSIFY_PROMPT.format(categories=list(EmailCategory)),
        response_model=EmailClassification,
    )
 
    # Step 2: 风险判断
    if result.needs_human or result.confidence < 0.7:
        # 低置信度或敏感类别 → 进入人工队列，不生成草稿
        queue_to_human(result)
        return
 
    # Step 3: 生成草稿（不发送）
    draft = create_draft(
        to=sender,
        subject=f"Re: {subject}",
        body=result.suggested_reply,
    )
 
    # Step 4: 通知客服审核草稿
    notify_review(draft_id=draft.id, classification=result)

# 内容分发工作流：单一数据源 + 多渠道适配
 
class ContentSource(BaseModel):
    """单一数据源：所有渠道版本必须从这里派生"""
    title: str
    body: str                    # Markdown 格式原文
    key_data: dict[str, str]     # 关键数据点（日期、价格、链接等）
    cta_url: str                 # 行动号召链接
 
class ChannelOutput(BaseModel):
    channel: str
    content: str
    key_data_preserved: bool     # 校验：关键数据是否保留
    word_count: int
    review_status: str = "pending"
 
CHANNEL_PROMPTS = {
    "wechat": """
将文章改写为微信公众号版本。
规则：
- 字数 1500-2500 字
- 段落短小，每段不超过 4 行
- 保留所有关键数据：{key_data}
- CTA 链接使用：{cta_url}
- 不使用 Markdown 语法（公众号编辑器不支持）
- 添加适当的小标题分隔
输出 JSON：{{"content": "...", "key_data_preserved": true/false}}
""",
    "twitter_thread": """
将文章改写为 Twitter/X 线程。
规则：
- 每条推文不超过 280 字符
- 总共 5-8 条
- 第一条是 hook，要吸引点击
- 最后一条包含 CTA 链接：{cta_url}
- 保留关键数据：{key_data}
输出 JSON：{{"tweets": [...], "key_data_preserved": true/false}}
""",
    "email": """
将文章改写为邮件订阅版本。
规则：
- 字数 800-1200 字
- 开头用一句话概括核心观点
- 保留所有关键数据：{key_data}
- CTA 按钮文案：「立即查看」，链接：{cta_url}
- 语气比公众号更私人化
输出 JSON：{{"content": "...", "key_data_preserved": true/false}}
""",
}
 
def distribute_content(source: ContentSource):
    outputs = []
 
    for channel, prompt_template in CHANNEL_PROMPTS.items():
        # Step 1: AI 生成渠道版本
        result = llm.generate(
            prompt=prompt_template.format(
                key_data=source.key_data,
                cta_url=source.cta_url,
            ),
            response_model=ChannelOutput,
        )
 
        # Step 2: 关键数据校验
        data_check = verify_key_data(
            content=result.content,
            expected_data=source.key_data,
        )
        if not data_check.passed:
            result.review_status = "data_mismatch"
            result.content = data_check.report  # 替换为差异报告
 
        # Step 3: 所有版本进入审核队列，不直接发布
        result.review_status = result.review_status or "pending_review"
        save_to_review_queue(channel, result)
        outputs.append(result)
 
    # Step 4: 生成跨渠道一致性报告
    consistency_report = check_cross_channel_consistency(outputs, source)
    return outputs, consistency_report

# ❌ 坏：直接用 AI 返回的文本，不校验结构
result = llm.chat("帮我分类这封邮件：" + email_body)
category = result  # result 可能是 "FAQ", "faq类", "这是一个FAQ问题" ...
if category == "FAQ":  # 永远匹配不上
    send_reply()

# ✅ 好：用 Pydantic 约束输出结构，枚举值写死
class Classification(BaseModel):
    category: Literal["faq", "billing", "technical", "complaint"]
    confidence: float = Field(ge=0, le=1)
 
result = llm.classify(prompt, response_model=Classification)
# result.category 一定是四个枚举值之一，confidence 一定是 0-1 之间的浮点数
if result.category == "faq" and result.confidence > 0.8:
    send_reply()

# ❌ 坏：直接覆盖整张表
sheet.clear()  # 清空所有数据
for row in new_data:
    sheet.append(row)  # 写入新数据
# 如果中途出错，旧数据已清空，新数据未写完 → 数据丢失

# ✅ 好：追加写入 + 幂等键 + 事务保护
existing_ids = set(sheet.get_column("order_id"))
new_rows = [r for r in new_data if r["order_id"] not in existing_ids]
 
# 批量写入，要么全部成功要么全部失败
sheet.batch_create(new_rows)
 
# 更新已有记录时，只更新变化的字段
for row in updated_data:
    sheet.batch_update(
        record_id=row["record_id"],
        fields={"status": row["status"]},  # 只更新 status，不动其他字段
    )

# ❌ 坏：catch 了异常什么都不做
try:
    result = llm.generate(prompt)
    send_email(result)
except Exception:
    pass  # 邮件没发出去，但没人知道

# ✅ 好：分类处理 + 重试 + 告警 + 降级
try:
    result = llm.generate(prompt, max_retries=3)
    create_draft(result)  # 生成草稿，不直接发送
except RateLimitError:
    # 限流：等待后重试，最多 3 次
    retry_with_backoff(lambda: llm.generate(prompt), max_retries=3)
except SchemaError as e:
    # 结构化输出格式错误：记录日志，进入人工队列
    logger.error(f"AI 输出格式错误: {e.detail}")
    queue_to_human_review(original_email)
except Exception as e:
    # 未知错误：告警 + 降级
    alert(f"邮件处理失败: {e}")
    queue_to_human_review(original_email)

# ❌ 坏：Prompt 太模糊，AI 自由发挥空间太大
prompt = f"请根据以下邮件内容生成回复：\n{email_body}"
# AI 可能写 500 字、用表情包、泄露内部信息、编造不存在的政策

# ✅ 好：明确约束边界，写出允许和禁止的具体规则
prompt = f"""
你是客服助手。根据邮件内容生成回复草稿。
 
约束：
- 字数：80-150 字
- 语气：专业友好，不使用表情包和网络用语
- 允许提及的信息：产品功能、文档链接、工单状态
- 禁止提及的信息：内部定价、未发布功能、其他客户信息
- 不确定的问题不要回答，改为建议客户联系客服
 
邮件内容：
{email_body}
 
输出 JSON：
{{"reply": "...", "uncertain_items": ["..."]}}
"""

# ❌ 坏：API Key 硬编码
API_KEY = "sk-abc123..."
SHEET_TOKEN = "bascn..."
 
# ✅ 好：从环境变量或密钥管理服务读取
import os
from aws_secretsmanager import get_secret
 
API_KEY = os.environ["LLM_API_KEY"]  # 或从 AWS Secrets Manager / Vault 读取
SHEET_TOKEN = get_secret("feishu/bitable/token")