当前位置：首页 > news >正文

Codex+Coze自动化工作流实战

news 2026/8/9 13:08:28

Codex（特指OpenAI的编程特化AI Agent）与Coze（扣子）平台的结合，能够实现从自然语言描述到可运行自动化流程的端到端生成。其核心在于利用Codex强大的代码理解和生成能力，来编写、调试并封装符合Coze平台规范的Skills（技能）或工作流节点脚本，最终在Coze环境中部署为可执行的自动化应用。下表概括了从启动到部署的核心环节：

阶段	核心目标	关键操作与工具	产出物
1. 环境准备	建立开发链路	配置Codex访问、获取Coze API Key、准备Python环境	可互联的开发环境
2. 需求解析与规划	将想法转化为结构化任务	使用Codex的“Plan Mode”或Threads进行任务拆解	详细的任务执行计划（JSON/伪代码）
3. 脚本开发与测试	实现各步骤功能	Codex生成Python代码，调用Coze SDK及第三方库	通过测试的、功能独立的Python模块
4. 封装与部署	将脚本集成到Coze平台	遵循Coze Skill规范打包，或在工作流编辑器中组装	已上线的Coze Skill或已发布的工作流

一、环境准备：连接Codex与Coze

这是自动化流程的基础，确保Codex生成的代码能在Coze平台上运行。

获取并配置Codex：你需要一个能访问Codex的环境。根据，Codex提供了多种入口：
- 官方桌面应用/网页版：功能最全，支持项目文件管理和Plan Mode。
- VS Code插件：在熟悉的IDE中集成。
- 命令行工具 (CLI)：适合自动化脚本集成。
  选择一种方式，确保你能与Codex进行自然语言交互以生成和运行代码。
准备Coze开发者凭证：
- 登录Coze平台，进入开发者中心或个人设置。
- 创建并获取你的API Key。这是Codex生成的脚本与Coze服务通信的凭证。

设置本地Python环境（可选但推荐）：
对于复杂的逻辑测试，在本地运行Codex生成的脚本会更高效。

# 创建虚拟环境 python -m venv coze_venv source coze_venv/bin/activate # Linux/Mac # coze_venv\Scripts\activate # Windows # 安装Coze官方SDK（如果提供）或必要的HTTP请求库 pip install requests python-dotenv

创建配置文件：
让Codex帮你创建一个配置文件来管理敏感信息。

# 文件名: config.py # Codex生成：用于存储Coze配置 import os from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 class CozeConfig: API_KEY = os.getenv("COZE_API_KEY", "你的实际API_KEY") API_BASE_URL = "https://api.coze.cn" # Coze API 基础地址 # 其他配置如工作空间ID等 WORKSPACE_ID = os.getenv("COZE_WORKSPACE_ID", "") # 在.env文件中填写： # COZE_API_KEY=your_api_key_here # COZE_WORKSPACE_ID=your_workspace_id

二、需求解析与规划：使用Codex拆解工作流

向Codex清晰描述你想要的工作流。例如：“创建一个每日早报工作流，它能从‘澎湃新闻’和‘虎嗅’的RSS抓取科技领域头条，用AI总结成一份不超过500字的简报，并生成配图建议，最后把结果保存到Notion数据库。”

指令示例：

“Codex，请为上述‘每日早报’需求创建一个详细的任务执行计划。将流程分解为可独立编码的步骤，并为每个步骤推荐合适的工具或库（优先使用Coze平台能力）。输出为JSON格式。”

Codex可能生成的计划（Plan Mode输出）：

{ "workflow_name": "DailyTechDigest", "steps": [ { "step_id": 1, "description": "从多个预设RSS源并发抓取最新文章", "implementation": "Python脚本，使用`feedparser`或`httpx`库。输出结构化的文章列表。", "coze_integration": "可封装为‘RSS阅读器’Skill，输入为URL列表，输出为文章JSON。" }, { "step_id": 2, "description": "过滤文章：只保留标题或摘要中含有关键词（如‘AI’、‘元宇宙’）的科技类文章", "implementation": "Python脚本，使用正则表达式或简单NLP库（如`jieba`）进行关键词匹配。", "coze_integration": "可作为代码节点嵌入工作流，或封装为‘内容过滤器’Skill。" }, { "step_id": 3, "description": "调用Coze平台的大模型能力，对筛选出的文章进行总结和简报撰写", "implementation": "通过Coze API（或SDK）调用Chat Completion，使用精心设计的提示词（Prompt）。", "coze_integration": "核心步骤。直接使用Coze平台的‘大模型’节点，或封装成‘智能摘要生成器’Skill。" }, { "step_id": 4, "description": "为简报生成配图建议（描述性文字）", "implementation": "同样调用Coze大模型，Prompt为‘根据以下简报内容，生成一段详细的配图描述...’。", "coze_integration": "与步骤3类似，使用大模型节点。" }, { "step_id": 5, "description": "将简报和配图建议写入指定的Notion数据库", "implementation": "Python脚本，使用Notion官方API (`notion-client`) 。", "coze_integration": "封装为‘Notion写入器’Skill，输入为标题、内容、标签等。" } ], "schedule": "每天上午7点自动触发", "trigger": "Coze工作流的‘定时触发器’节点" }

这个结构化的计划是后续编码的蓝图。

三、脚本开发与测试：Codex生成功能代码

基于上述计划，指导Codex为每个步骤生成可运行的Python代码。以**步骤1（抓取RSS）和步骤3（调用Coze AI总结）**为例：

# 文件名: rss_fetcher.py # Codex生成：步骤1 - 多源RSS抓取 import feedparser import asyncio import aiohttp from typing import List, Dict import json async def fetch_single_feed(session: aiohttp.ClientSession, url: str) -> List[Dict]: """异步抓取单个RSS源""" try: async with session.get(url, timeout=10) as response: xml_data = await response.text() feed = feedparser.parse(xml_data) articles = [] for entry in feed.entries[:10]: # 每个源取最新10条 article = { "title": entry.title, "link": entry.link, "published": entry.get("published", ""), "source": url, "summary": entry.get("summary", ""), "content": entry.get("content", [{}])[0].get("value", "") if entry.get("content") else "" } articles.append(article) return articles except Exception as e: print(f"抓取 {url} 失败: {e}") return [] async def fetch_all_feeds(rss_urls: List[str]) -> List[Dict]: """并发抓取所有RSS源""" async with aiohttp.ClientSession() as session: tasks = [fetch_single_feed(session, url) for url in rss_urls] results = await asyncio.gather(*tasks) # 扁平化结果列表 all_articles = [article for sublist in results for article in sublist] return all_articles # 测试代码 if __name__ == "__main__": import asyncio urls = [ "https://rsshub.app/thepaper/featured", # 澎湃新闻 "https://rsshub.app/huxiu/tag/291" # 虎嗅-人工智能 ] articles = asyncio.run(fetch_all_feeds(urls)) print(f"共抓取到 {len(articles)} 篇文章") print(json.dumps(articles[:1], indent=2, ensure_ascii=False)) # 打印第一篇作为示例

# 文件名: coze_summarizer.py # Codex生成：步骤3 - 调用Coze API生成摘要 import requests import json from config import CozeConfig # 导入之前的配置 def generate_digest_with_coze(articles: List[Dict], system_prompt: str) -> str: """ 调用Coze的对话API，根据多篇文章生成一份简报。 Args: articles: 文章列表，每个元素是包含title, link, summary等的字典。 system_prompt: 定义AI角色和任务的系统提示词。 Returns: 生成的简报文本。 """ # 1. 构建用户消息：将文章信息整理成文本 articles_text = " --- ".join([ f"标题：{a['title']} 来源：{a['source']} 摘要：{a['summary'][:200]}..." for a in articles[:5] # 限制前5篇，防止token超限 ]) user_message = f"请基于以下科技新闻，生成一份简洁的每日早报： {articles_text}" # 2. 准备API请求载荷 url = f"{CozeConfig.API_BASE_URL}/v1/chat/completions" # 假设的API端点 headers = { "Authorization": f"Bearer {CozeConfig.API_KEY}", "Content-Type": "application/json" } payload = { "model": "coze-model-pro", # 指定使用的模型 "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message} ], "temperature": 0.7, "max_tokens": 800 } # 3. 发送请求 try: response = requests.post(url, headers=headers, data=json.dumps(payload), timeout=30) response.raise_for_status() result = response.json() return result["choices"][0]["message"]["content"] except requests.exceptions.RequestException as e: return f"API请求失败: {e}" except KeyError as e: return f"解析API响应失败: {e}" # 测试代码 if __name__ == "__main__": # 模拟一些文章数据 test_articles = [ {"title": "GPT-5发布在即，性能提升引期待", "source": "澎湃新闻", "summary": "OpenAI宣布下一代模型..."}, {"title": "AI芯片竞争白热化", "source": "虎嗅", "summary": "国内外多家公司发布新品..."} ] system_prompt = """你是一位专业的科技媒体编辑。请将提供的新闻条目整合成一份流畅、精炼的每日科技简报。 要求：语言简洁明了，突出核心进展，字数控制在500字以内。""" digest = generate_digest_with_coze(test_articles, system_prompt) print("生成的简报：") print(digest)

让Codex运行并调试这些脚本，直到每个功能模块都能独立正常工作。

四、封装与部署：在Coze平台上创建工作流

这是将Codex生成的代码“产品化”的最后一步。你有两种主要路径：

路径A：封装为Coze Skill（推荐用于功能复用）
Skill是Coze平台的可复用能力模块。将上述脚本打包成Skill。

创建Skill项目结构：

daily_digest_skill/ ├── SKILL.md # Skill描述文档（必须） ├── main.py # 主执行文件 ├── requirements.txt # Python依赖 └── config.json # 技能配置（可选）

编写SKILL.md（让Codex根据功能生成）：

# 每日科技简报生成器 ## 描述 自动从多个RSS源抓取科技新闻，通过大模型生成一份简洁的每日简报。 ## 输入参数 * `rss_urls` (字符串数组): RSS源地址列表。 * `system_prompt` (字符串，可选): 定制AI简报生成角色的提示词。 * `max_articles_per_feed` (整数，可选，默认5): 每个源抓取的最大文章数。 ## 输出 * `digest` (字符串): 生成的每日简报文本。 * `processed_articles` (整数): 实际处理的文章数量。 ## 使用方法 在工作流中添加本Skill，连接输入输出即可。

编写符合Coze Skill接口的main.py（让Codex适配）：

# main.py - 适配Coze Skill调用规范 import sys import json import asyncio # 假设将之前的函数放在lib目录下 from lib.rss_fetcher import fetch_all_feeds from lib.coze_summarizer import generate_digest_with_coze def main(input_data: dict) -> dict: """Skill主入口函数，Coze平台会调用此函数""" # 1. 解析输入 rss_urls = input_data.get("rss_urls", []) system_prompt = input_data.get("system_prompt", "你是一位科技编辑...") max_articles = input_data.get("max_articles_per_feed", 5) if not rss_urls: return {"error": "请输入至少一个RSS URL"} # 2. 执行核心逻辑（注意异步函数处理） try: # 抓取文章 loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) all_articles = loop.run_until_complete(fetch_all_feeds(rss_urls)) loop.close() # 生成简报 digest = generate_digest_with_coze(all_articles[:max_articles*len(rss_urls)], system_prompt) # 3. 返回输出 return { "digest": digest, "processed_articles": len(all_articles), "status": "success" } except Exception as e: return {"error": str(e), "status": "fail"} # 用于本地测试 if __name__ == "__main__": test_input = { "rss_urls": ["https://rsshub.app/thepaper/featured"], "system_prompt": "生成简短摘要。" } result = main(test_input) print(json.dumps(result, indent=2, ensure_ascii=False))

打包与上传：将整个目录打包为ZIP文件，在Coze平台的“技能库” -> “创建技能”页面，选择“从ZIP包上传”，即可创建属于你自己的Skill。

路径B：在Coze工作流编辑器中可视化组装
对于一次性或逻辑复杂、需要多条件分支的工作流，直接在Coze画布上拖拽节点更直观。

创建新工作流：在Coze工作室点击“创建”->“工作流”。
添加并配置节点：
- 触发器：从左侧添加“定时触发器”，设置为每天7:00。
- 代码节点：将Codex生成并测试好的**步骤2（过滤文章）**的Python代码粘贴进去。代码节点可以直接处理上游数据。
- 大模型节点：添加“大模型”节点，将步骤3和4的提示词逻辑配置进去。你可以直接把之前在coze_summarizer.py中调优好的system_prompt和user_message模板复制过来。
- Skill节点：如果你已将**步骤5（写入Notion）**封装成了Skill，直接在这里搜索并添加“Notion写入器”Skill，并配置数据库ID等信息。
- 判断/循环节点：根据需要添加逻辑控制。
连接节点并测试：用连接线将节点按顺序链接起来。点击“测试运行”，检查数据是否在各个节点间正确流转。
发布：测试成功后，保存并发布工作流。发布后，该工作流可以单独运行，也可以被其他Bot当作一个技能来调用。

关键配置与最佳实践

API密钥管理：永远不要在Codex对话或生成的代码中硬编码API Key。使用环境变量（.env文件）或Coze平台提供的密钥管理功能。
错误处理与日志：指导Codex在生成的代码中加入完善的try-except块和日志记录，便于在Coze工作流运行失败时排查问题。
Token与成本控制：在调用Coze大模型API的步骤中，通过Codex设置合理的max_tokens参数，并对输入文本进行截断，以控制成本。
增量开发与测试：不要试图一次性让Codex生成整个复杂工作流。采用“分步生成-分步测试”的策略，先让单个节点（或Skill）跑通，再逐步组装。
利用Codex的Plan Mode：对于复杂项目，充分利用Codex的“Plan Mode”功能，让它先输出详细的设计方案，你再基于此方案要求它生成各部分代码，这样结构更清晰。

通过以上方法，你可以有效地将Codex作为“高级开发助手”，驱动在Coze平台上从零到一自动创作出功能完整、可维护的工作流。其本质是将Codex的代码生成能力与Coze的低代码编排和部署能力相结合，实现“所想即所得”的自动化应用开发。