用 Python 与 Claude 搭建文本清洗流水线

本项目展示了如何构建一条异步自动化流水线：使用 Python 脚本处理非结构化文本数据（例如客户评价、支持工单或原始文本日志），通过 n8n 编排流程，并借助 Claude 3.5 Sonnet（Anthropic）增强处理能力。

这套方案将人工文本分类的工作量减少了 90%，输出干净、结构化的 JSON 数据，可直接写入数据库或供下游分析仪表盘使用。

自动文本规范化：清理空白字符，处理特殊字符，并验证数据结构。

AI 驱动的实体与情感提取：利用 Claude 3.5 Sonnet 对反馈进行分类，判断情感倾向（Positive、Negative、Neutral），并评估紧急程度（High、Medium、Low）。

基于规则的回退引擎：当 API 或 webhook 无法访问时，使用本地正则表达式和关键词解析，保障服务的高可用性。

n8n 工作流集成：将不同的数据接入渠道直接连接到 LLM 处理节点。

🛠️ 架构与技术栈

编程语言：Python 3.10+（pandas、requests、python-dotenv）

编排工具：n8n（自托管或 n8n Cloud）

LLM 引擎：通过 Anthropic API 使用 Claude 3.5 Sonnet / Claude 3 Haiku

数据格式：JSON、CSV、Markdown

版本控制：Git / GitHub

[ 非结构化数据（CSV/JSON） ] │ ▼ [ Python 预处理 ] │ ▼ [ 入站 Webhook（n8n） ] │ ▼ [ 处理与推理（Claude AI） ] │ ▼ [ 结构化输出（干净的 JSON） ]

本地环境已安装 Python 3.10+。

一个正在运行的 n8n 实例。

一个 Anthropic API key（ANTHROPIC_API_KEY）。

已配置的 Python 虚拟环境（venv）。

💻 第 1 步：使用 Python 提取和预处理数据

下面的脚本读取包含原始客户反馈的 CSV 文件，执行基础文本规范化（去除特殊字符和空白字符），然后将数据发送到 n8n webhook。

import pandas as pd
import requests
import json
import re

WEBHOOK_URL = "https://your-n8n-instance.cloud/webhook/process-data"

def clean_text(text: str) -> str:
    """Sanitizes text by stripping line breaks, tab characters, and extra spaces."""
    if not isinstance(text, str):
        return ""
    text = re.sub(r'[\r\n\t]+', ' ', text)
    text = re.sub(r'\s+', ' ', text)
    return text.strip()

def process_and_send(file_path: str):
    """Loads CSV data, cleans input strings, and posts payload to n8n webhook."""
    df = pd.read_csv(file_path)

    # Assuming input column is named 'feedback'
    df['clean_feedback'] = df['feedback'].apply(clean_text)

    payload = {
        "source": "customer_feedback_batch",
        "records": df[['id', 'clean_feedback']].to_dict(orient='records')
    }

    response = requests.post(
        WEBHOOK_URL, 
        headers={"Content-Type": "application/json"},
        data=json.dumps(payload)
    )

    if response.status_code == 200:
        print("✅ Data successfully transmitted to n8n.")
    else:
        print(f"❌ Transmission failed: {response.status_code} - {response.text}")

if __name__ == "__main__":
    process_and_send("input_data.csv")

⚙️ 第 2 步：配置 n8n 工作流

n8n 工作流由四个核心节点组成：

Webhook Node：监听 HTTP POST 请求，接收 Python 发送的数据。

Item Lists / Split Node：拆分 records 数组，独立处理每条文本。

HTTP Request / LangChain Node（Claude AI）：使用严格要求 JSON 输出的 system prompt，将文本转发给 Anthropic API。

Respond to Webhook / Storage Node：汇总结构化输出。

用于 Claude AI 的 System Prompt：

You are a senior data analyst specializing in text classification.
Analyze the following customer feedback and return ONLY a valid JSON object adhering strictly to this schema:

{
  "sentiment": "POSITIVE" | "NEUTRAL" | "NEGATIVE",
  "category": "TECHNICAL_SUPPORT" | "BILLING" | "USABILITY" | "OTHER",
  "urgency_score": 1-5,
  "summary": "Concise single-sentence summary"
}

Feedback input: {{ $json.clean_feedback }}

📊 第 3 步：预期输出

自动化工作流返回的处理结果：

{
  "id": 1042,
  "original_text": "The app crashes every time I try to download my billing statement in PDF.",
  "analysis": {
    "sentiment": "NEGATIVE",
    "category": "TECHNICAL_SUPPORT",
    "urgency_score": 4,
    "summary": "Critical crash occurring during PDF billing statement downloads inside the mobile application."
  }
}

🧪 测试与错误处理

API 超时：如果 Claude API 的响应时间超过 10 秒，n8n 会触发失败自动重试策略，采用指数退避，最多尝试 3 次。

JSON 验证：如果 Claude 在 JSON 数据之外返回了意料之外的对话文本，Code 节点中的正则表达式会提取 {...} 括号包裹的内容，确保下游能够正常解析。

姓名：Marcela Zapata Vanegas

专长：技术写作、数据分析与 AI 自动化

GitHub：github.com/whoismarce

如需采取进一步措施，可以考虑屏蔽此人和/或举报滥用行为。