API自动化Python工作流
脚本编写
详细介绍如何通过 Claude API 构建自动化工作流，包括批量处理、文本分类、内容生成流水线、与 n8n/Zapier 集成，以及多步骤 AI 任务编排的实际代码示例。
· 阅读约 16 分钟
Claude API 不只是用来对话——当你把它嵌入自动化流程，真正的效率提升才开始。本文从实际场景出发，介绍如何构建 AI 驱动的工作流。
自动化工作流的核心思路
把 Claude 当成工作流中的一个处理节点：
输入数据 → [预处理] → [Claude 处理] → [后处理] → 输出结果
Claude 擅长处理的环节：
- 文本分类和打标签
- 内容提取和结构化
- 内容生成和改写
- 质量判断和过滤
- 摘要和转换
环境准备
# 安装依赖
pip install anthropic python-dotenv
# .env 文件
ANTHROPIC_API_KEY=sk-ant-...
import anthropic
import os
from dotenv import load_dotenv
load_dotenv()
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
场景一：批量文本处理
批量文章分类
import anthropic
import json
client = anthropic.Anthropic()
def classify_article(title: str, content: str) -> dict:
"""将文章分类到预定义类别"""
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=256,
messages=[{
"role": "user",
"content": f"""请将以下文章分类，返回 JSON 格式：
标题：{title}
内容前200字：{content[:200]}
返回格式：
{{
"category": "主分类（科技/财经/娱乐/体育/其他）",
"subcategory": "子分类",
"confidence": 0.95,
"keywords": ["关键词1", "关键词2", "关键词3"]
}}
只返回 JSON，不要其他内容。"""
}]
)
result_text = message.content[0].text
return json.loads(result_text)
# 批量处理
articles = [
{"title": "苹果发布新款 MacBook", "content": "苹果公司今日..."},
{"title": "A股午后跳水", "content": "沪深两市..."},
]
results = []
for article in articles:
result = classify_article(article["title"], article["content"])
results.append({**article, **result})
print(f"✓ {article['title']} → {result['category']}")
print(json.dumps(results, ensure_ascii=False, indent=2))
批量情感分析
def analyze_sentiment_batch(reviews: list[str]) -> list[dict]:
"""批量分析用户评论情感"""
# 一次 API 调用处理多条（节省调用次数）
reviews_text = "\n".join([f"{i+1}. {r}" for i, r in enumerate(reviews)])
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{
"role": "user",
"content": f"""分析以下 {len(reviews)} 条用户评论的情感，返回 JSON 数组：
{reviews_text}
返回格式（数组顺序必须和输入一致）：
[
{{
"sentiment": "positive/negative/neutral",
"score": 0.85,
"reason": "一句话说明"
}}
]
只返回 JSON 数组。"""
}]
)
return json.loads(message.content[0].text)
# 使用示例
reviews = [
"产品很好，包装精美，快递也快！",
"质量太差了，用了两天就坏了，差评！",
"中规中矩，还可以吧",
]
results = analyze_sentiment_batch(reviews)
for review, result in zip(reviews, results):
print(f"{result['sentiment']:10} | {review[:20]}...")
场景二：内容生成流水线
关键词到文章的自动化流程
from dataclasses import dataclass
@dataclass
class ArticleGenerationResult:
keyword: str
outline: str
article: str
meta_description: str
word_count: int
def generate_article_pipeline(keyword: str) -> ArticleGenerationResult:
"""
完整文章生成流水线：
关键词 → 大纲 → 文章 → Meta
"""
# 步骤1：生成大纲
print(f" [1/3] 生成大纲...")
outline_response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=512,
messages=[{
"role": "user",
"content": f"为关键词「{keyword}」生成一篇SEO友好文章的大纲。"
f"包含H1、3-4个H2、每个H2下2个H3。只输出大纲结构。"
}]
)
outline = outline_response.content[0].text
# 步骤2：基于大纲生成文章
print(f" [2/3] 生成文章...")
article_response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
messages=[{
"role": "user",
"content": f"根据以下大纲，写一篇关于「{keyword}」的完整文章，"
f"约1500字，内容充实，有实用价值：\n\n{outline}"
}]
)
article = article_response.content[0].text
# 步骤3：生成 Meta Description
print(f" [3/3] 生成 Meta...")
meta_response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=256,
messages=[{
"role": "user",
"content": f"为以下文章写一个SEO优化的Meta Description（120-155字符）：\n\n{article[:500]}"
}]
)
meta = meta_response.content[0].text.strip()
return ArticleGenerationResult(
keyword=keyword,
outline=outline,
article=article,
meta_description=meta,
word_count=len(article)
)
# 批量运行
keywords = ["Claude 使用技巧", "AI 写作助手", "提示词工程"]
for keyword in keywords:
print(f"\n处理: {keyword}")
result = generate_article_pipeline(keyword)
# 保存到文件
with open(f"output/{keyword}.md", "w", encoding="utf-8") as f:
f.write(f"---\n")
f.write(f"title: {result.outline.split('\\n')[0]}\n")
f.write(f"description: {result.meta_description}\n")
f.write(f"---\n\n")
f.write(result.article)
print(f" ✓ 完成，{result.word_count} 字")
场景三：数据提取和结构化
从非结构化文本提取结构化数据
def extract_job_info(job_posting: str) -> dict:
"""从招聘信息中提取结构化数据"""
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=512,
messages=[{
"role": "user",
"content": f"""从以下招聘信息中提取结构化数据，返回JSON：
{job_posting}
提取字段：
- company: 公司名称
- position: 职位名称
- location: 工作地点
- salary_min: 最低薪资（数字，万/年）
- salary_max: 最高薪资（数字，万/年）
- experience_years: 要求工作年限（数字）
- education: 学历要求
- skills: 技能要求（数组）
- benefits: 福利待遇（数组）
- remote: 是否支持远程（true/false/hybrid）
字段缺失时返回 null，只输出JSON。"""
}]
)
return json.loads(message.content[0].text)
# 处理多个招聘信息
import glob
job_files = glob.glob("jobs/*.txt")
structured_jobs = []
for file_path in job_files:
with open(file_path, encoding="utf-8") as f:
content = f.read()
job_data = extract_job_info(content)
job_data["source_file"] = file_path
structured_jobs.append(job_data)
# 导出为 CSV
import pandas as pd
df = pd.DataFrame(structured_jobs)
df.to_csv("jobs_structured.csv", index=False, encoding="utf-8-sig")
print(f"处理完成，共 {len(structured_jobs)} 条记录")
场景四：多轮对话自动化
自动化审查流程（AI 自我检查）
def self_review_pipeline(content: str, content_type: str) -> dict:
"""
内容自我审查流水线：
生成 → 审查 → 改进 → 最终版本
"""
conversation = []
# 回合1：初始生成
conversation.append({
"role": "user",
"content": f"请写一段{content_type}：{content}"
})
first_draft = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=conversation
)
draft_text = first_draft.content[0].text
conversation.append({"role": "assistant", "content": draft_text})
# 回合2：让 AI 审查自己的输出
conversation.append({
"role": "user",
"content": "请审查上面你写的内容，指出3个最主要的问题（逻辑/表达/完整性）。"
})
review = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=512,
messages=conversation
)
review_text = review.content[0].text
conversation.append({"role": "assistant", "content": review_text})
# 回合3：根据审查改进
conversation.append({
"role": "user",
"content": "根据上面的审查意见，给出改进后的最终版本。"
})
final = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=conversation
)
final_text = final.content[0].text
return {
"original": draft_text,
"review": review_text,
"final": final_text
}
场景五：与 n8n 集成
n8n 是开源的工作流自动化工具（类似 Zapier，但可自托管）。
在 n8n 中调用 Claude API
在 n8n 中添加 HTTP Request 节点：
{
"method": "POST",
"url": "https://api.anthropic.com/v1/messages",
"headers": {
"x-api-key": "{{ $credentials.anthropicApi.apiKey }}",
"anthropic-version": "2023-06-01",
"content-type": "application/json"
},
"body": {
"model": "claude-sonnet-4-6",
"max_tokens": 1024,
"messages": [
{
"role": "user",
"content": "{{ $json.inputText }}"
}
]
}
}
常用 n8n + Claude 工作流
工作流1：邮件自动分类和回复草稿
Gmail 触发 → HTTP Request (Claude 分类) → 条件判断 →
[投诉] → Claude 生成回复草稿 → Gmail 草稿
[咨询] → Claude 查找 FAQ → Gmail 自动回复
[其他] → 标签并归档
工作流2：社交媒体内容生产
定时触发（每天9点）→ RSS 读取热点 →
Claude 生成社交媒体文案 → Buffer/小红书 API → 发布
工作流3：客服工单处理
新工单触发 → Claude 分析优先级和类型 →
[紧急] → Slack 通知相关人员
[常规] → Claude 生成回复建议 → 人工确认 → 发送
性能优化和最佳实践
并发处理（速度提升 5-10x）
import asyncio
import anthropic
async_client = anthropic.AsyncAnthropic()
async def process_single(text: str) -> str:
message = await async_client.messages.create(
model="claude-haiku-4-5-20251001", # 简单任务用 Haiku 更快更便宜
max_tokens=256,
messages=[{"role": "user", "content": text}]
)
return message.content[0].text
async def process_batch(texts: list[str]) -> list[str]:
tasks = [process_single(text) for text in texts]
return await asyncio.gather(*tasks)
# 运行
texts = ["文本1", "文本2", "文本3", "..."]
results = asyncio.run(process_batch(texts))
模型选择原则
| 任务 | 推荐模型 | 原因 |
|---|---|---|
| 简单分类/提取 | claude-haiku-4-5-20251001 | 速度快，价格低 |
| 日常生成/分析 | claude-sonnet-4-6 | 质量和性价比最佳 |
| 复杂推理/长文 | claude-opus-4-7 | 最高质量 |
错误处理和重试
import time
from anthropic import RateLimitError, APIError
def call_with_retry(prompt: str, max_retries: int = 3) -> str:
for attempt in range(max_retries):
try:
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": prompt}]
)
return message.content[0].text
except RateLimitError:
wait_time = 2 ** attempt # 指数退避: 1s, 2s, 4s
print(f"频率限制，等待 {wait_time}s 后重试...")
time.sleep(wait_time)
except APIError as e:
print(f"API 错误: {e}")
if attempt == max_retries - 1:
raise
raise Exception("达到最大重试次数")
成本控制
# 在调用前估算 token 数（避免意外超支）
def estimate_cost(input_text: str, model: str = "claude-sonnet-4-6") -> dict:
# 粗略估算：1 token ≈ 1.5 个中文字符 或 4 个英文字符
char_count = len(input_text)
estimated_input_tokens = char_count // 2 # 保守估算
# claude-sonnet-4-6 价格（2025年参考）
prices = {
"claude-haiku-4-5-20251001": {"input": 0.25, "output": 1.25},
"claude-sonnet-4-6": {"input": 3, "output": 15},
"claude-opus-4-7": {"input": 15, "output": 75},
}
price = prices.get(model, prices["claude-sonnet-4-6"])
input_cost = estimated_input_tokens * price["input"] / 1_000_000
return {
"estimated_tokens": estimated_input_tokens,
"estimated_cost_usd": round(input_cost, 6),
}
标记本节教程为已读
记录您的学习进度，方便后续查看。