把AI工具串成自动流程:从配置文件到日报生成的完整实战教程
AI工具工作流自动化教程|附配置文件
在过去几年里,AI工具已经从“单点能力”逐渐演变为“工作流引擎”。早期我们使用AI,通常是打开一个聊天窗口,输入需求,复制结果,再手动粘贴到文档、表格、邮件或代码仓库中。这个过程虽然提高了效率,但仍然存在大量重复操作:整理资料、生成摘要、分类信息、发送通知、同步数据、输出报告等。
真正能够释放AI生产力的方式,不是单独使用某一个AI工具,而是把AI嵌入到完整的工作流中,让它自动接收输入、理解任务、调用工具、生成结果,并把结果传递到下一个环节。
本文将以“AI工具工作流自动化”为主题,系统讲解如何搭建一个可复用、可扩展的自动化流程,并附上可参考的配置文件示例。无论你是内容运营、产品经理、开发者、市场人员,还是希望提升个人效率的知识工作者,都可以基于本文思路搭建自己的AI自动化系统。
一、什么是AI工作流自动化?
AI工作流自动化,简单来说,就是将多个任务步骤串联起来,并让AI参与其中的关键决策或内容生成环节。
一个典型的AI自动化工作流可能包括:
- 监听某个输入来源,例如表单、邮件、RSS、数据库、Webhook;
- 提取输入数据,例如标题、正文、链接、用户需求;
- 调用AI模型进行处理,例如总结、分类、润色、翻译、生成方案;
- 根据AI返回结果执行动作,例如写入数据库、发送通知、创建任务;
- 输出最终结果,例如生成日报、文章、邮件、工单或知识库文档。
举个例子:
当用户提交一个产品反馈表单后,系统自动调用AI分析反馈类型,判断是Bug、功能建议还是使用咨询;然后自动生成摘要,分配优先级,并创建到项目管理工具中,同时通知对应负责人。
在这个过程中,人不需要反复复制粘贴,也不需要手动判断每条反馈的类别。AI承担了理解和初步处理的工作,而自动化平台负责流程编排。
二、适合自动化的AI工具场景
并不是所有工作都适合自动化。一般来说,适合引入AI自动化的场景通常具备以下特点:
- 重复性强:每天或每周都会发生类似任务;
- 输入结构相对稳定:例如邮件、表单、文档、客服记录;
- 输出格式有规律:例如摘要、分类、标签、报告;
- 人工判断成本较高但标准明确;
- 需要跨工具流转数据。
以下是几个常见应用场景。
1. 内容生产自动化
适合自媒体、运营、市场团队使用。
例如:
- 从RSS自动抓取行业新闻;
- 调用AI生成中文摘要;
- 根据摘要生成小红书、公众号、微博版本;
- 自动存入Notion或飞书文档;
- 每天定时发送到群聊。
2. 客服与工单自动化
适合SaaS公司、电商、在线教育等业务。
例如:
- 收集用户反馈;
- AI识别用户意图;
- 判断问题紧急程度;
- 自动生成回复建议;
- 创建工单并分配给对应团队。
3. 销售线索处理自动化
适合B2B业务、咨询业务、课程转化等场景。
例如:
- 用户提交咨询表单;
- AI分析客户行业、预算、需求;
- 自动给销售线索打分;
- 同步到CRM;
- 给销售发送跟进建议。
4. 个人知识管理自动化
适合研究人员、学生、产品经理、咨询顾问。
例如:
- 收藏网页链接;
- AI自动总结核心观点;
- 提取关键词;
- 写入知识库;
- 定期生成学习周报。
5. 代码与研发辅助自动化
适合开发团队。
例如:
- Git提交后自动生成变更说明;
- Issue创建时自动分类;
- PR提交后AI生成Review建议;
- 每日自动汇总项目进度;
- 根据错误日志生成排查建议。
三、AI工作流自动化的基本架构
搭建AI工作流之前,我们需要理解它的基本组成。一个完整的AI自动化系统通常包含五个部分。
1. 触发器 Trigger
触发器决定工作流什么时候开始执行。
常见触发方式包括:
- 定时触发:每天8点生成日报;
- Webhook触发:收到表单提交后立即执行;
- 文件触发:网盘新增文件后开始处理;
- 邮件触发:收到指定邮件后自动分析;
- 数据库触发:新增一条记录后启动流程;
- 手动触发:点击按钮后执行。
2. 数据输入 Input
输入是AI处理的原始材料,例如:
- 用户提交的文字;
- 文档内容;
- 网页链接;
- 表格数据;
- 聊天记录;
- 商品评论;
- 错误日志。
输入质量越高,AI输出越稳定。因此在流程中通常需要进行数据清洗,例如去除HTML标签、截断无关内容、统一字段格式。
3. AI处理节点 AI Node
这是整个流程的核心。
AI节点可以执行多种任务:
- 总结;
- 分类;
- 翻译;
- 改写;
- 提取结构化信息;
- 生成标题;
- 判断优先级;
- 输出JSON;
- 生成代码;
- 生成邮件回复。
为了让AI输出稳定,推荐使用结构化Prompt,并明确输出格式。
4. 条件判断 Condition
AI返回结果后,流程通常需要根据结果进入不同分支。
例如:
- 如果优先级为高,则立即发送通知;
- 如果类型为Bug,则创建研发工单;
- 如果类型为咨询,则发送给销售;
- 如果情绪为负面,则标记为重点关注;
- 如果内容不合规,则进入人工审核。
5. 执行动作 Action
最后一步是把处理结果发送到目标系统。
常见动作包括:
- 写入Notion、飞书、多维表格;
- 发送企业微信、钉钉、Slack消息;
- 创建Jira、Trello、Linear任务;
- 写入MySQL、PostgreSQL、MongoDB;
- 发送邮件;
- 生成Markdown文档;
- 调用其他API。
四、推荐工具组合
不同用户可以根据技术能力选择不同方案。
方案一:低代码自动化工具
适合非技术用户。
常见工具包括:
- Zapier;
- Make;
- n8n;
- Pipedream;
- 飞书集成平台;
- 钉钉宜搭;
- 腾讯云自动化助手。
优点:
- 上手快;
- 可视化配置;
- 适合业务人员;
- 不需要写太多代码。
缺点:
- 复杂逻辑受限;
- 大规模使用成本较高;
- 对私有化部署支持有限。
方案二:n8n + AI API
适合希望兼顾灵活性和成本的用户。
n8n是一个开源自动化工作流平台,支持本地部署,也支持云端运行。它可以连接Webhook、数据库、HTTP API、邮件、表格等多种服务,非常适合搭建AI工作流。
方案三:Python脚本 + 定时任务
适合开发者。
使用Python可以完全自定义流程,例如:
- 使用
requests调用AI API; - 使用
schedule或cron定时执行; - 使用
pandas处理表格; - 使用数据库保存状态;
- 使用企业微信机器人发送通知。
优点是灵活、可控、成本低;缺点是需要编程基础。
五、实战案例:自动生成行业资讯日报
下面我们设计一个完整的AI工作流案例:
每天早上8点,系统自动抓取指定RSS源中的最新文章,调用AI生成中文摘要和价值判断,再整理成Markdown日报,最后发送到企业微信群,并同步保存到本地文件。
这个案例适用于:
- 运营团队追踪行业动态;
- 投资研究人员监控新闻;
- 产品经理关注竞品变化;
- 个人学习者整理信息流。
六、流程设计
整体流程如下:
定时触发
↓
读取RSS源
↓
筛选最近24小时文章
↓
提取标题、链接、发布时间、摘要
↓
调用AI生成中文解读
↓
整理为Markdown日报
↓
保存到本地文件
↓
发送到企业微信机器人
输入数据示例
RSS文章原始数据可能包含以下字段:
{
"title": "OpenAI launches new enterprise features",
"link": "https://example.com/news/openai-enterprise",
"published": "2026-06-03T02:30:00Z",
"summary": "OpenAI introduced several new features for enterprise customers..."
}
AI期望输出示例
我们希望AI返回结构化内容:
{
"title_cn": "OpenAI推出新的企业级功能",
"summary_cn": "OpenAI发布了面向企业客户的新功能,重点提升安全、权限管理和团队协作能力。",
"importance": "高",
"category": "AI产品",
"insight": "企业级AI工具正在从单一模型能力转向权限、安全和组织协作能力竞争。"
}
七、项目目录结构
如果使用Python实现,可以采用如下目录:
ai-news-workflow/
├── config.yaml
├── prompt.md
├── main.py
├── requirements.txt
├── output/
│ └── daily-report.md
└── logs/
└── app.log
各文件作用如下:
| 文件 | 作用 |
|---|---|
config.yaml |
存放RSS源、AI模型、企业微信Webhook等配置 |
prompt.md |
存放AI提示词模板 |
main.py |
主程序 |
requirements.txt |
Python依赖 |
output/ |
保存生成的日报 |
logs/ |
保存运行日志 |
八、配置文件示例:config.yaml
下面是一个可参考的配置文件。
app:
name: "AI行业资讯日报工作流"
timezone: "Asia/Shanghai"
output_dir: "./output"
max_articles: 10
schedule:
enabled: true
run_time: "08:00"
rss:
sources:
- name: "OpenAI Blog"
url: "https://openai.com/blog/rss.xml"
- name: "Google AI Blog"
url: "https://blog.google/technology/ai/rss/"
- name: "The Verge AI"
url: "https://www.theverge.com/rss/ai-artificial-intelligence/index.xml"
ai:
provider: "openai-compatible"
base_url: "https://api.example.com/v1"
api_key: "${AI_API_KEY}"
model: "gpt-4o-mini"
temperature: 0.3
max_tokens: 1200
prompt:
file: "./prompt.md"
output_format: "json"
wechat:
enabled: true
webhook_url: "${WECHAT_WEBHOOK_URL}"
mention_all: false
logging:
level: "INFO"
file: "./logs/app.log"
配置说明
app.max_articles:限制每天处理的文章数量,避免API调用过多;rss.sources:可以添加多个RSS源;ai.base_url:如果使用兼容OpenAI格式的服务,可以在这里填写对应地址;ai.api_key:建议通过环境变量读取,不要直接写死;wechat.webhook_url:企业微信机器人地址;prompt.file:提示词模板文件路径。
九、提示词模板:prompt.md
为了让AI输出稳定,建议将提示词单独管理。
你是一名专业的AI行业分析师,擅长将英文科技新闻转化为中文行业简报。
请根据以下文章信息,生成结构化JSON结果。
文章标题:
{{title}}
文章来源:
{{source}}
发布时间:
{{published}}
文章摘要:
{{summary}}
文章链接:
{{link}}
请完成以下任务:
1. 将标题翻译成自然流畅的中文;
2. 用不超过120字总结文章核心内容;
3. 判断重要性,只能从「高」「中」「低」中选择;
4. 判断分类,只能从以下类别中选择:
- AI产品
- 模型发布
- 投融资
- 行业应用
- 政策监管
- 开源项目
- 其他
5. 给出一句对行业或业务的启发。
请严格返回JSON,不要输出Markdown,不要添加解释。
返回格式如下:
{
"title_cn": "",
"summary_cn": "",
"importance": "",
"category": "",
"insight": ""
}
这个提示词有几个关键点:
- 明确AI角色;
- 明确输入字段;
- 明确任务步骤;
- 限制分类选项;
- 强制JSON输出;
- 避免额外解释。
这些细节可以显著提高自动化流程的稳定性。
十、Python依赖文件:requirements.txt
feedparser==6.0.11
requests==2.32.3
PyYAML==6.0.2
python-dotenv==1.0.1
schedule==1.2.2
如果你还需要更复杂的时间处理,可以额外安装:
python-dateutil==2.9.0
十一、主程序示例:main.py
下面是一份简化版代码,便于理解流程。实际生产环境中建议加入更完善的异常处理、重试机制、数据库去重和日志系统。
import os
import json
import time
import yaml
import requests
import feedparser
import schedule
from datetime import datetime, timedelta, timezone
from dotenv import load_dotenv
load_dotenv()
def load_config(path="config.yaml"):
with open(path, "r", encoding="utf-8") as f:
raw = f.read()
raw = raw.replace("${AI_API_KEY}", os.getenv("AI_API_KEY", ""))
raw = raw.replace("${WECHAT_WEBHOOK_URL}", os.getenv("WECHAT_WEBHOOK_URL", ""))
return yaml.safe_load(raw)
def load_prompt_template(path):
with open(path, "r", encoding="utf-8") as f:
return f.read()
def render_prompt(template, article):
prompt = template
for key, value in article.items():
prompt = prompt.replace("{{" + key + "}}", str(value or ""))
return prompt
def fetch_rss_articles(config):
articles = []
now = datetime.now(timezone.utc)
since = now - timedelta(hours=24)
for source in config["rss"]["sources"]:
feed = feedparser.parse(source["url"])
for entry in feed.entries:
published = None
if hasattr(entry, "published_parsed") and entry.published_parsed:
published = datetime(*entry.published_parsed[:6], tzinfo=timezone.utc)
if published and published < since:
continue
articles.append({
"source": source["name"],
"title": entry.get("title", ""),
"link": entry.get("link", ""),
"published": published.isoformat() if published else "",
"summary": entry.get("summary", "")
})
return articles[:config["app"]["max_articles"]]
def call_ai(config, prompt):
ai_config = config["ai"]
url = ai_config["base_url"].rstrip("/") + "/chat/completions"
headers = {
"Authorization": f"Bearer {ai_config['api_key']}",
"Content-Type": "application/json"
}
payload = {
"model": ai_config["model"],
"temperature": ai_config.get("temperature", 0.3),
"max_tokens": ai_config.get("max_tokens", 1200),
"messages": [
{
"role": "system",
"content": "你是一个严谨的结构化信息处理助手。"
},
{
"role": "user",
"content": prompt
}
]
}
response = requests.post(url, headers=headers, json=payload, timeout=60)
response.raise_for_status()
content = response.json()["choices"][0]["message"]["content"]
try:
return json.loads(content)
except json.JSONDecodeError:
return {
"title_cn": "解析失败",
"summary_cn": content[:200],
"importance": "低",
"category": "其他",
"insight": "AI返回结果不是合法JSON,需要优化提示词或增加解析容错。"
}
def generate_markdown_report(results):
today = datetime.now().strftime("%Y-%m-%d")
lines = []
lines.append(f"# AI行业资讯日报|{today}")
lines.append("")
lines.append("以下内容由AI自动抓取、总结和整理,仅供参考。")
lines.append("")
high_items = [r for r in results if r["ai"].get("importance") == "高"]
if high_items:
lines.append("## 今日重点")
lines.append("")
for item in high_items:
ai = item["ai"]
lines.append(f"- **{ai['title_cn']}**:{ai['summary_cn']}")
lines.append("")
lines.append("## 资讯列表")
lines.append("")
for idx, item in enumerate(results, 1):
ai = item["ai"]
raw = item["raw"]
lines.append(f"### {idx}. {ai.get('title_cn', raw['title'])}")
lines.append("")
lines.append(f"- 来源:{raw['source']}")
lines.append(f"- 分类:{ai.get('category', '其他')}")
lines.append(f"- 重要性:{ai.get('importance', '低')}")
lines.append(f"- 链接:{raw['link']}")
lines.append("")
lines.append(f"**摘要:** {ai.get('summary_cn', '')}")
lines.append("")
lines.append(f"**启发:** {ai.get('insight', '')}")
lines.append("")
return "\n".join(lines)
def save_report(config, markdown):
output_dir = config["app"]["output_dir"]
os.makedirs(output_dir, exist_ok=True)
filename = datetime.now().strftime("daily-report-%Y-%m-%d.md")
path = os.path.join(output_dir, filename)
with open(path, "w", encoding="utf-8") as f:
f.write(markdown)
return path
def send_wechat(config, markdown):
if not config["wechat"].get("enabled"):
return
webhook_url = config["wechat"]["webhook_url"]
if not webhook_url:
print("未配置企业微信Webhook,跳过发送。")
return
content = markdown
# 企业微信机器人文本消息有长度限制,生产环境建议分段发送
if len(content) > 3500:
content = content[:3500] + "\n\n内容过长,已截断,请查看完整日报文件。"
payload = {
"msgtype": "markdown",
"markdown": {
"content": content
}
}
response = requests.post(webhook_url, json=payload, timeout=30)
response.raise_for_status()
def run_workflow():
config = load_config()
prompt_template = load_prompt_template(config["prompt"]["file"])
print("开始抓取RSS文章...")
articles = fetch_rss_articles(config)
print(f"获取到 {len(articles)} 篇文章。")
results = []
for article in articles:
print(f"正在处理:{article['title']}")
prompt = render_prompt(prompt_template, article)
ai_result = call_ai(config, prompt)
results.append({
"raw": article,
"ai": ai_result
})
time.sleep(1)
markdown = generate_markdown_report(results)
path = save_report(config, markdown)
send_wechat(config, markdown)
print(f"日报生成完成:{path}")
if __name__ == "__main__":
config = load_config()
if config["schedule"].get("enabled"):
run_time = config["schedule"]["run_time"]
schedule.every().day.at(run_time).do(run_workflow)
print(f"工作流已启动,将在每天 {run_time} 执行。")
while True:
schedule.run_pending()
time.sleep(30)
else:
run_workflow()
十二、环境变量配置:.env
为了安全起见,不建议把API Key直接写入配置文件。可以新建一个.env文件:
AI_API_KEY=你的AI服务API密钥
WECHAT_WEBHOOK_URL=你的企业微信机器人Webhook地址
同时建议将.env加入.gitignore:
.env
logs/
output/
这样可以避免敏感信息被上传到代码仓库。
十三、如何运行项目?
第一步:创建项目目录
mkdir ai-news-workflow
cd ai-news-workflow
第二步:安装依赖
pip install -r requirements.txt
第三步:配置环境变量
cp .env.example .env
然后填写你的API Key和Webhook地址。
第四步:测试运行
如果想立即运行一次,可以把config.yaml中的定时任务关闭:
schedule:
enabled: false
run_time: "08:00"
然后执行:
python main.py
如果控制台输出“日报生成完成”,并且output目录下出现Markdown文件,就说明流程已经跑通。
十四、工作流优化建议
基础流程跑通以后,可以进一步优化。
1. 增加去重机制
RSS文章可能重复出现,建议使用链接作为唯一标识,将已处理链接写入数据库或本地文件。
可以使用简单的JSON文件:
{
"processed_links": [
"https://example.com/news/001",
"https://example.com/news/002"
]
}
每次处理前先判断链接是否存在,避免重复消耗AI Token。
2. 增加失败重试
调用AI API、RSS源或企业微信Webhook时,可能因为网络原因失败。生产环境应加入重试机制,例如最多重试3次,每次间隔5秒。
3. 使用JSON Schema约束输出
如果AI服务支持结构化输出,可以使用JSON Schema约束字段类型,进一步减少解析失败。
4. 增加人工审核节点
对于需要公开发布的内容,不建议完全自动发布。更稳妥的做法是:
- AI生成初稿;
- 自动保存到文档;
- 通知编辑审核;
- 编辑确认后再发布。
5. 增加成本控制
可以从以下方面降低成本:
- 限制文章数量;
- 先用规则过滤无关内容;
- 对短文本使用小模型;
- 对重要内容再调用强模型;
- 缓存AI处理结果;
- 设置每日Token预算。
十五、n8n配置思路示例
如果你不想写代码,也可以使用n8n搭建类似流程。
节点可以这样设计:
Cron Trigger
↓
RSS Feed Read
↓
Split In Batches
↓
OpenAI Chat Model
↓
Code Node解析JSON
↓
Markdown模板拼接
↓
HTTP Request发送企业微信
↓
Write Binary File保存日报
n8n中AI节点Prompt示例
你是一名AI行业资讯分析师。
请将以下资讯整理为JSON:
标题:{{$json["title"]}}
链接:{{$json["link"]}}
摘要:{{$json["contentSnippet"]}}
返回字段:
- title_cn
- summary_cn
- importance
- category
- insight
要求:
1. importance只能是高、中、低;
2. category只能是AI产品、模型发布、投融资、行业应用、政策监管、开源项目、其他;
3. 只返回JSON。
n8n配置导入片段示例
以下不是完整导出文件,但可以作为配置思路:
{
"name": "AI News Daily Workflow",
"nodes": [
{
"parameters": {
"triggerTimes": {
"item": [
{
"hour": 8,
"minute": 0
}
]
}
},
"name": "Daily Cron",
"type": "n8n-nodes-base.cron",
"typeVersion": 1,
"position": [240, 300]
},
{
"parameters": {
"url": "https://openai.com/blog/rss.xml"
},
"name": "Read RSS",
"type": "n8n-nodes-base.rssFeedRead",
"typeVersion": 1,
"position": [460, 300]
},
{
"parameters": {
"method": "POST",
"url": "https://api.example.com/v1/chat/completions",
"sendHeaders": true,
"headerParameters": {
"parameters": [
{
"name": "Authorization",
"value": "Bearer {{$env.AI_API_KEY}}"
},
{
"name": "Content-Type",
"value": "application/json"
}
]
},
"sendBody": true,
"bodyParametersJson": "{ \"model\": \"gpt-4o-mini\", \"messages\": [{ \"role\": \"user\", \"content\": \"请总结以下资讯并返回JSON:{{$json.title}} {{$json.contentSnippet}}\" }] }"
},
"name": "Call AI API",
"type": "n8n-nodes-base.httpRequest",
"typeVersion": 4,
"position": [700, 300]
}
],
"connections": {
"Daily Cron": {
"main": [
[
{
"node": "Read RSS",
"type": "main",
"index": 0
}
]
]
},
"Read RSS": {
"main": [
[
{
"node": "Call AI API",
"type": "main",
"index": 0
}
]
]
}
}
}
如果你使用n8n生产环境,建议将API Key保存在Credentials或环境变量中,不要直接写在节点参数里。
十六、常见问题与解决方案
问题一:AI返回的不是合法JSON怎么办?
解决方法:
- 在Prompt中明确“只返回JSON”;
- 给出完整JSON示例;
- 降低
temperature; - 增加解析容错;
- 使用支持结构化输出的模型接口;
- 在失败后让AI进行一次“JSON修复”。
问题二:日报内容质量不稳定怎么办?
可以优化输入和提示词:
- 给AI提供更完整的文章摘要;
- 增加角色设定;
- 明确读者对象;
- 限制输出长度;
- 给出优质示例;
- 对重要文章使用更强模型。
问题三:自动化流程会不会误发内容?
会有风险。因此涉及外部发布、客户回复、财务审批、法律文本等场景时,建议保留人工审核环节。AI适合生成建议和初稿,不应在高风险场景中完全替代人工决策。
问题四:如何降低API成本?
可以采取分层处理:
- 先用规则过滤无关数据;
- 用小模型做初筛;
- 只对高价值内容调用强模型;
- 对相同输入做缓存;
- 限制每日处理数量。
十七、从单一流程到自动化系统
当你完成第一个AI工作流后,可以继续扩展为完整系统。例如:
- 资讯日报;
- 竞品监控;
- 用户反馈分析;
- 销售线索评分;
- 客服回复建议;
- 会议纪要整理;
- 周报自动生成;
- 项目风险提醒。
这些流程看似不同,但底层结构非常相似:
输入数据 → 清洗数据 → AI理解/生成 → 条件判断 → 执行动作 → 保存结果
因此,真正重要的不是某一个工具,而是建立“流程化思维”。只要你能把工作拆解为清晰步骤,就可以逐步把其中重复、标准化、耗时的部分交给AI和自动化系统完成。
十八、总结
AI工具工作流自动化的核心价值,不是让AI替你完成所有工作,而是让AI承担重复、繁琐、标准化的信息处理任务,让人把精力放在判断、创造和决策上。
本文介绍了AI自动化工作流的基本概念、适用场景、系统架构、工具选择,并以“AI行业资讯日报”为例,提供了完整的配置文件、提示词模板、Python代码和n8n配置思路。
如果你是初学者,建议从一个小流程开始,例如“自动总结收藏文章”或“自动生成每日简报”;如果你是团队负责人,可以优先选择高频、重复、低风险的业务环节试点,例如反馈分类、日报生成、线索整理等。
搭建AI自动化系统的关键原则可以概括为五句话:
- 先流程,后工具:先想清楚业务步骤,再选择平台;
- 先小闭环,后大系统:从一个可运行的小流程开始;
- 先辅助,后自动:高风险场景保留人工审核;
- 先稳定,后复杂:输出格式和异常处理比炫技更重要;
- 先复用,后扩展:把Prompt、配置和代码模块化。
当AI不再只是聊天窗口,而是成为工作流中的一个处理节点时,效率提升才会真正发生。你可以从本文提供的配置文件开始,改成自己的RSS源、业务字段、通知渠道和提示词模板,逐步搭建属于自己的AI自动化工作系统。