AiToEarn 这类 AI 智能体内容自动化项目要解决的核心问题不是“怎么调用大模型接口”而是“内容从选题、写作、审核到发布这条链路如何被可靠地编排和运维”。很多开发者的第一版实现是把所有要求写进一个提示词让模型一次性输出成稿结果内容同质化严重、缺少审核环节、发布仍然要人工拷贝。本文以 AiToEarn 为原型讲清楚如何用四个 AI 智能体跑通内容全链路选题策划智能体负责产出结构化内容简报内容创作智能体负责把简报扩写成正文编辑审核智能体负责质量门禁发布运营智能体负责把通过审核的内容投递到目标平台并回传数据。适合准备搭建多智能体内容工作流、正在做 AI 智能体落地选型、或者想在现有系统中引入内容自动化能力的开发者阅读。按文中骨架可以复现一条最小可运行的智能体流水线再逐步补齐日志、重试、成本和合规控制。1. AiToEarn 是什么四个智能体如何串成一条内容流水线1.1 四个智能体的分工和边界AiToEarn 的链路设计核心是“每个智能体只干一件事”。不需要让一个大模型同时完成选题、写作、审核和发布而是把内容生产过程拆成四个独立环节每个环节由一个智能体负责上游输出成为下游输入。四个智能体的分工如下智能体输入输出核心职责选题策划智能体 Research用户主题、关键词、目标平台内容简报 JSON确定选题角度、标题候选、大纲、目标字数内容创作智能体 Writer内容简报正文初稿 JSON按简报生成标题、摘要和完整正文编辑审核智能体 Editor正文初稿审核结论 JSON检查结构、逻辑、关键词和合规风险发布运营智能体 Publisher审核通过的稿件发布状态 JSON平台投递、链接回填、数据回传边界要划清楚。Research 只做规划和判断不写正文Writer 只负责生成不判断内容能不能发布Editor 只做审核不改写全文Publisher 只负责发布动作和结果记录不重新组织内容。这样做的好处是每个环节的提示词都很简单输出结构稳定一旦出错可以准确定位到具体智能体。1.2 数据在链路中的流转方式整条链路的数据流可以抽象成下面这条带方向的管线用户输入主题 - Research: 内容简报 brief - Writer: 正文初稿 draft - Editor: 审核结论 review - Publisher: 发布状态 publication每两个智能体之间传递的是结构化 JSON而不是一段自然语言。结构化数据让每个环节都有明确的输入输出边界方便记录、回放和断点续跑。比如 Editor 审核不通过时流程可以带着审核意见重新回到 Writer 修改而不是把未通过的内容继续往后送。实际项目中还要考虑审核分支。AiToEarn 的典型回退路径是Editor 输出passed: false后流程回到 Writer同时把 issues 写入下一轮 Writer 的输入让模型按问题清单修改。回退次数必须有限制否则一个长期不过审的稿件会反复消耗 Token。1.3 多智能体编排与单次提示词生成的区别有的项目里开发者用一条超长提示词让模型“既当编辑又当作者”一次调用直接生成成品。这种做法在快速验证时可行但进入工程化之后会遇到几个问题提示词里的角色指令互相冲突模型不知道当前到底该以哪个身份输出中间状态不可控无法在写作完成之后、发布之前插入人工审核某一处出错只能整体重来。多智能体编排解决了这些问题同时也有代价。维度单次提示词生成多智能体编排提示词复杂度高角色容易冲突低每个智能体一个角色中间状态可控性低只能看到最终结果高每个环节都有独立产物局部重试不支持失败整体重来支持只重跑失败环节Token 消耗低只调用一次高全链路多次调用排查难度难定位问题环节按环节日志快速定位智能体工程里常说的“把智能体约束在可控流程中”本质上就是这个思路模型负责单步推理智能体负责把多步推理组织成可验证、可回退的流程。AiToEarn 采用四智能体结构正是为了让这种约束更清晰。2. 先把工程骨架搭起来环境、依赖和配置2.1 运行环境与依赖清单动手之前先确认运行环境。本文示例使用 Python 3.11 及以上版本通过 OpenAI 兼容接口调用大模型。如果实际项目使用其他模型服务商只要接口兼容代码结构可以保持不变SDK 版本和模型名称需要按服务商文档调整。组件版本建议用途Python3.11 及以上运行环境openai1.x调用大模型聊天补全接口PyYAML6.x读取智能体和流程配置python-dotenv1.x从 .env 文件读取密钥pydantic2.x定义消息结构并做校验创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate pip install openai pyyaml python-dotenv pydantic这里不要把 API Key 直接写在代码里。项目根目录放一个.env.example模板真实密钥放到本地的.env文件并在.gitignore中忽略它# .env.example OPENAI_API_KEYsk-your-key OPENAI_BASE_URLhttps://api.example.com/v1注意密钥只通过环境变量或密钥管理服务注入。生产环境建议使用云厂商的密钥管理工具并开启定期轮换不要提交到代码仓库。2.2 项目目录结构一个适合本文场景的目录结构如下aitoearn/ ├── config/ │ ├── agents.yaml │ └── .env.example ├── aitoearn/ │ ├── core/ │ │ ├── __init__.py │ │ ├── message.py │ │ ├── agent.py │ │ ├── llm.py │ │ └── workflow.py │ ├── agents/ │ │ ├── __init__.py │ │ ├── research.py │ │ ├── writer.py │ │ ├── editor.py │ │ └── publisher.py │ └── main.py ├── output/ │ ├── briefs/ │ ├── drafts/ │ └── published/ └── requirements.txtcore目录放与业务无关的通用能力消息对象、智能体基类、模型调用封装、编排器。agents目录放四个具体智能体。output目录保存每一轮产生的中间产物方便回放和排查。2.3 统一配置与智能体参数智能体参数统一放在config/agents.yaml而不是散落在代码里defaults: timeout: 120 retry: 3 models: research: gpt-4o-mini writer: gpt-4o editor: gpt-4o-mini publisher: gpt-4o-mini workflow: steps: - research - writer - editor - publisher max_review_rounds: 2这里有两个关键点。第一不同环节使用不同模型和参数Writer 承担长文生成使用能力更强的模型和更大的max_tokensEditor 只需要做判断用小模型和更低的temperature就够。第二max_review_rounds控制 Editor 打回重写的最大轮数防止稿件反复不过审导致流程死循环。3. 设计四个智能体的输入输出协议3.1 选题策划智能体输出内容简报Research 的输入是用户给的宽泛主题、关键词和目标平台输出是一份结构化内容简报。简报字段越固定后续 Writer 的生成质量越稳定。{ brief_id: 20250101-001, title_candidates: [标题A, 标题B, 标题C], angle: 从智能体编排角度讲内容自动化, outline: [ 为什么需要多智能体, 四个智能体的分工, 编排器如何串联, 生产环境注意事项 ], keywords: [AI智能体, 工作流搭建, 内容自动化], target_word_count: 5000, platform: tech_blog }title_candidates提供多个标题候选避免直接绑定一个可能不够吸引人的标题outline是后续 Writer 生成正文的骨架target_word_count用于约束正文长度。在提示词里要明确要求“只输出 JSON 对象不要输出多余解释”否则模型可能把思考过程一起输出导致解析失败。3.2 内容创作智能体按简报生成正文Writer 的唯一输入是 Research 输出的简报输出是正文初稿。它不需要思考选题是否合理也不需要判断内容是否合规只需要把简报变成完整文章。{ draft_id: 20250101-001, brief_id: 20250101-001, title: 标题A, summary: 用四个AI智能体串起内容生产全链路的技术实践, content: # 正文 Markdown 内容……, seo_keywords: [AI智能体, 工作流搭建] }正文使用 Markdown 格式存储方便后续发布到支持 Markdown 的平台也方便在编辑阶段做结构解析。brief_id用于关联上游简报保证数据血缘可追踪。3.3 编辑审核智能体质量门禁Editor 是链路里的质量闸口。它检查初稿结构是否完整、逻辑是否连贯、关键词是否自然分布、是否存在明显风险表达然后输出审核结论。{ review_id: 20250101-001, draft_id: 20250101-001, passed: false, score: 62, issues: [ { type: structure, description: 正文缺少结尾总结段 }, { type: keyword, description: 核心关键词只出现1次密度偏低 } ], suggestions: [ 补充结尾总结段, 在开头自然加入核心关键词 ] }passed字段直接决定流程往哪个方向走通过则进入 Publisher不通过则回到 Writer 重写。issues和suggestions要一起回传给 Writer否则模型不知道具体该改哪里。Editor 的temperature建议调低避免审核结论时好时坏。3.4 发布运营智能体多平台投递与数据回传Publisher 接收审核通过的稿件负责发布动作和结果记录。真实接入平台前先设计一个符合预期的发布结果结构{ publication_id: 20250101-001, platform: tech_blog, status: published, url: https://example.com/articles/xxx, published_at: 2025-01-01T10:30:0008:00, metrics: { views: 0, likes: 0 } }把真实平台发布当成一个适配器问题处理不同平台有不同的鉴权方式、内容规范和接口协议统一实现一个PlatformAdapter接口每个平台写一个适配器。发布动作建议加人工审批闸口或 dry-run 模式避免未经确认就对外发布。演示项目可以先模拟发布把结果写入本地output/published目录等接入官方 API 时再替换适配器实现。4. 用编排器把四个智能体串起来4.1 定义统一消息对象链路里的每个环节都传递同一个消息对象核心字段包括当前步骤、载荷、状态、错误信息和元数据。# aitoearn/core/message.py from dataclasses import dataclass, field from typing import Any, Optional dataclass class Message: step: str payload: dict status: str pending error: Optional[str] None metadata: dict field(default_factorydict) def to_json(self): return { step: self.step, payload: self.payload, status: self.status, error: self.error, metadata: self.metadata, }step表示当前消息应该交给哪个智能体也用来表达流程下一步往哪里走。payload携带业务数据随着流程推进不断累积。metadata记录来源、任务 ID、轮次等辅助信息不参与业务生成但排错时非常关键。4.2 编写智能体基类和通用模型调用所有智能体继承同一个基类基类负责异常包装和状态更新# aitoearn/core/agent.py import logging from abc import ABC, abstractmethod from .message import Message logger logging.getLogger(__name__) class AgentError(Exception): pass class BaseAgent(ABC): name: str base def __init__(self, client, model: str, temperature: float 0.5, max_tokens: int 2048): self.client client self.model model self.temperature temperature self.max_tokens max_tokens abstractmethod def process(self, message: Message) - Message: raise NotImplementedError def run(self, message: Message) - Message: try: result self.process(message) result.status done return result except Exception as exc: logger.exception(agent %s processing failed., self.name) message.status failed message.error str(exc) raise AgentError(str(exc)) from exc模型调用统一封装成函数并尽量要求模型返回 JSON 对象减少解析失败概率# aitoearn/core/llm.py import json import re def chat_json(client, model, system_prompt, user_payload, temperature0.5, max_tokens2048): response client.chat.completions.create( modelmodel, temperaturetemperature, max_tokensmax_tokens, messages[ {role: system, content: system_prompt}, {role: user, content: json.dumps(user_payload, ensure_asciiFalse)}, ], response_format{type: json_object}, ) return parse_model_json(response.choices[0].message.content) def parse_model_json(content: str) - dict: try: return json.loads(content) except json.JSONDecodeError: match re.search(r(?:json)?\s*(.*?)\s*, content, re.DOTALL) if match: return json.loads(match.group(1)) raiseparse_model_json先尝试直接解析解析失败再尝试从 Markdown 代码块中提取 JSON。这个兜底逻辑能处理一部分模型输出格式不稳定的情况。需要注意的是response_format参数并非所有模型和兼容端点都支持接入新服务商时先确认能力再启用。4.3 实现策划、创作、审核三个智能体Research 智能体的实现示例# aitoearn/agents/research.py from ..core.agent import BaseAgent from ..core.llm import chat_json from ..core.message import Message SYSTEM_PROMPT 你是一个内容选题策划。 根据用户输入的主题和关键词输出结构化内容简报。 简报必须包含 title_candidates、angle、outline、keywords、target_word_count、platform。 只输出 JSON 对象不要输出多余解释。 class ResearchAgent(BaseAgent): name research def process(self, message: Message) - Message: payload message.payload brief chat_json( self.client, self.model, SYSTEM_PROMPT, payload, temperatureself.temperature, max_tokensself.max_tokens, ) return Message( stepwriter, payload{brief: brief, task_id: payload.get(task_id)}, metadata{source: research, brief_id: brief.get(brief_id)}, )Writer 和 Editor 的结构类似。Editor 的关键在于根据审核结果决定下一步走向# aitoearn/agents/editor.py from ..core.agent import BaseAgent from ..core.llm import chat_json from ..core.message import Message SYSTEM_PROMPT 你是内容审核编辑。 检查稿件是否结构完整、逻辑连贯、关键词合理、内容合规。 输出 passed、score、issues、suggestions 四个字段。 只输出 JSON 对象。 class EditorAgent(BaseAgent): name editor def process(self, message: Message) - Message: payload message.payload review chat_json( self.client, self.model, SYSTEM_PROMPT, payload, temperature0.2, max_tokens1024, ) passed bool(review.get(passed)) return Message( steppublisher if passed else writer, payload{**payload, review: review}, metadata{source: editor, passed: passed}, )step字段在这里充当路由信息审核通过进入publisher不通过回到writer。Editor 的temperature固定为 0.2保证审核结论尽量稳定。4.4 发布智能体与合规边界Publisher 在演示环境先不做真实网络请求而是把发布结果写入本地记录# aitoearn/agents/publisher.py from
AiToEarn 这类 AI 智能体内容自动化项目要解决的核心问题不是“怎么调用大模型接口”而是“内容从选题、写作、审核到发布这条链路如何被可靠地编排和运维”。很多开发者的第一版实现是把所有要求写进一个提示词让模型一次性输出成稿结果内容同质化严重、缺少审核环节、发布仍然要人工拷贝。本文以 AiToEarn 为原型讲清楚如何用四个 AI 智能体跑通内容全链路选题策划智能体负责产出结构化内容简报内容创作智能体负责把简报扩写成正文编辑审核智能体负责质量门禁发布运营智能体负责把通过审核的内容投递到目标平台并回传数据。适合准备搭建多智能体内容工作流、正在做 AI 智能体落地选型、或者想在现有系统中引入内容自动化能力的开发者阅读。按文中骨架可以复现一条最小可运行的智能体流水线再逐步补齐日志、重试、成本和合规控制。1. AiToEarn 是什么四个智能体如何串成一条内容流水线1.1 四个智能体的分工和边界AiToEarn 的链路设计核心是“每个智能体只干一件事”。不需要让一个大模型同时完成选题、写作、审核和发布而是把内容生产过程拆成四个独立环节每个环节由一个智能体负责上游输出成为下游输入。四个智能体的分工如下智能体输入输出核心职责选题策划智能体 Research用户主题、关键词、目标平台内容简报 JSON确定选题角度、标题候选、大纲、目标字数内容创作智能体 Writer内容简报正文初稿 JSON按简报生成标题、摘要和完整正文编辑审核智能体 Editor正文初稿审核结论 JSON检查结构、逻辑、关键词和合规风险发布运营智能体 Publisher审核通过的稿件发布状态 JSON平台投递、链接回填、数据回传边界要划清楚。Research 只做规划和判断不写正文Writer 只负责生成不判断内容能不能发布Editor 只做审核不改写全文Publisher 只负责发布动作和结果记录不重新组织内容。这样做的好处是每个环节的提示词都很简单输出结构稳定一旦出错可以准确定位到具体智能体。1.2 数据在链路中的流转方式整条链路的数据流可以抽象成下面这条带方向的管线用户输入主题 - Research: 内容简报 brief - Writer: 正文初稿 draft - Editor: 审核结论 review - Publisher: 发布状态 publication每两个智能体之间传递的是结构化 JSON而不是一段自然语言。结构化数据让每个环节都有明确的输入输出边界方便记录、回放和断点续跑。比如 Editor 审核不通过时流程可以带着审核意见重新回到 Writer 修改而不是把未通过的内容继续往后送。实际项目中还要考虑审核分支。AiToEarn 的典型回退路径是Editor 输出passed: false后流程回到 Writer同时把 issues 写入下一轮 Writer 的输入让模型按问题清单修改。回退次数必须有限制否则一个长期不过审的稿件会反复消耗 Token。1.3 多智能体编排与单次提示词生成的区别有的项目里开发者用一条超长提示词让模型“既当编辑又当作者”一次调用直接生成成品。这种做法在快速验证时可行但进入工程化之后会遇到几个问题提示词里的角色指令互相冲突模型不知道当前到底该以哪个身份输出中间状态不可控无法在写作完成之后、发布之前插入人工审核某一处出错只能整体重来。多智能体编排解决了这些问题同时也有代价。维度单次提示词生成多智能体编排提示词复杂度高角色容易冲突低每个智能体一个角色中间状态可控性低只能看到最终结果高每个环节都有独立产物局部重试不支持失败整体重来支持只重跑失败环节Token 消耗低只调用一次高全链路多次调用排查难度难定位问题环节按环节日志快速定位智能体工程里常说的“把智能体约束在可控流程中”本质上就是这个思路模型负责单步推理智能体负责把多步推理组织成可验证、可回退的流程。AiToEarn 采用四智能体结构正是为了让这种约束更清晰。2. 先把工程骨架搭起来环境、依赖和配置2.1 运行环境与依赖清单动手之前先确认运行环境。本文示例使用 Python 3.11 及以上版本通过 OpenAI 兼容接口调用大模型。如果实际项目使用其他模型服务商只要接口兼容代码结构可以保持不变SDK 版本和模型名称需要按服务商文档调整。组件版本建议用途Python3.11 及以上运行环境openai1.x调用大模型聊天补全接口PyYAML6.x读取智能体和流程配置python-dotenv1.x从 .env 文件读取密钥pydantic2.x定义消息结构并做校验创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate pip install openai pyyaml python-dotenv pydantic这里不要把 API Key 直接写在代码里。项目根目录放一个.env.example模板真实密钥放到本地的.env文件并在.gitignore中忽略它# .env.example OPENAI_API_KEYsk-your-key OPENAI_BASE_URLhttps://api.example.com/v1注意密钥只通过环境变量或密钥管理服务注入。生产环境建议使用云厂商的密钥管理工具并开启定期轮换不要提交到代码仓库。2.2 项目目录结构一个适合本文场景的目录结构如下aitoearn/ ├── config/ │ ├── agents.yaml │ └── .env.example ├── aitoearn/ │ ├── core/ │ │ ├── __init__.py │ │ ├── message.py │ │ ├── agent.py │ │ ├── llm.py │ │ └── workflow.py │ ├── agents/ │ │ ├── __init__.py │ │ ├── research.py │ │ ├── writer.py │ │ ├── editor.py │ │ └── publisher.py │ └── main.py ├── output/ │ ├── briefs/ │ ├── drafts/ │ └── published/ └── requirements.txtcore目录放与业务无关的通用能力消息对象、智能体基类、模型调用封装、编排器。agents目录放四个具体智能体。output目录保存每一轮产生的中间产物方便回放和排查。2.3 统一配置与智能体参数智能体参数统一放在config/agents.yaml而不是散落在代码里defaults: timeout: 120 retry: 3 models: research: gpt-4o-mini writer: gpt-4o editor: gpt-4o-mini publisher: gpt-4o-mini workflow: steps: - research - writer - editor - publisher max_review_rounds: 2这里有两个关键点。第一不同环节使用不同模型和参数Writer 承担长文生成使用能力更强的模型和更大的max_tokensEditor 只需要做判断用小模型和更低的temperature就够。第二max_review_rounds控制 Editor 打回重写的最大轮数防止稿件反复不过审导致流程死循环。3. 设计四个智能体的输入输出协议3.1 选题策划智能体输出内容简报Research 的输入是用户给的宽泛主题、关键词和目标平台输出是一份结构化内容简报。简报字段越固定后续 Writer 的生成质量越稳定。{ brief_id: 20250101-001, title_candidates: [标题A, 标题B, 标题C], angle: 从智能体编排角度讲内容自动化, outline: [ 为什么需要多智能体, 四个智能体的分工, 编排器如何串联, 生产环境注意事项 ], keywords: [AI智能体, 工作流搭建, 内容自动化], target_word_count: 5000, platform: tech_blog }title_candidates提供多个标题候选避免直接绑定一个可能不够吸引人的标题outline是后续 Writer 生成正文的骨架target_word_count用于约束正文长度。在提示词里要明确要求“只输出 JSON 对象不要输出多余解释”否则模型可能把思考过程一起输出导致解析失败。3.2 内容创作智能体按简报生成正文Writer 的唯一输入是 Research 输出的简报输出是正文初稿。它不需要思考选题是否合理也不需要判断内容是否合规只需要把简报变成完整文章。{ draft_id: 20250101-001, brief_id: 20250101-001, title: 标题A, summary: 用四个AI智能体串起内容生产全链路的技术实践, content: # 正文 Markdown 内容……, seo_keywords: [AI智能体, 工作流搭建] }正文使用 Markdown 格式存储方便后续发布到支持 Markdown 的平台也方便在编辑阶段做结构解析。brief_id用于关联上游简报保证数据血缘可追踪。3.3 编辑审核智能体质量门禁Editor 是链路里的质量闸口。它检查初稿结构是否完整、逻辑是否连贯、关键词是否自然分布、是否存在明显风险表达然后输出审核结论。{ review_id: 20250101-001, draft_id: 20250101-001, passed: false, score: 62, issues: [ { type: structure, description: 正文缺少结尾总结段 }, { type: keyword, description: 核心关键词只出现1次密度偏低 } ], suggestions: [ 补充结尾总结段, 在开头自然加入核心关键词 ] }passed字段直接决定流程往哪个方向走通过则进入 Publisher不通过则回到 Writer 重写。issues和suggestions要一起回传给 Writer否则模型不知道具体该改哪里。Editor 的temperature建议调低避免审核结论时好时坏。3.4 发布运营智能体多平台投递与数据回传Publisher 接收审核通过的稿件负责发布动作和结果记录。真实接入平台前先设计一个符合预期的发布结果结构{ publication_id: 20250101-001, platform: tech_blog, status: published, url: https://example.com/articles/xxx, published_at: 2025-01-01T10:30:0008:00, metrics: { views: 0, likes: 0 } }把真实平台发布当成一个适配器问题处理不同平台有不同的鉴权方式、内容规范和接口协议统一实现一个PlatformAdapter接口每个平台写一个适配器。发布动作建议加人工审批闸口或 dry-run 模式避免未经确认就对外发布。演示项目可以先模拟发布把结果写入本地output/published目录等接入官方 API 时再替换适配器实现。4. 用编排器把四个智能体串起来4.1 定义统一消息对象链路里的每个环节都传递同一个消息对象核心字段包括当前步骤、载荷、状态、错误信息和元数据。# aitoearn/core/message.py from dataclasses import dataclass, field from typing import Any, Optional dataclass class Message: step: str payload: dict status: str pending error: Optional[str] None metadata: dict field(default_factorydict) def to_json(self): return { step: self.step, payload: self.payload, status: self.status, error: self.error, metadata: self.metadata, }step表示当前消息应该交给哪个智能体也用来表达流程下一步往哪里走。payload携带业务数据随着流程推进不断累积。metadata记录来源、任务 ID、轮次等辅助信息不参与业务生成但排错时非常关键。4.2 编写智能体基类和通用模型调用所有智能体继承同一个基类基类负责异常包装和状态更新# aitoearn/core/agent.py import logging from abc import ABC, abstractmethod from .message import Message logger logging.getLogger(__name__) class AgentError(Exception): pass class BaseAgent(ABC): name: str base def __init__(self, client, model: str, temperature: float 0.5, max_tokens: int 2048): self.client client self.model model self.temperature temperature self.max_tokens max_tokens abstractmethod def process(self, message: Message) - Message: raise NotImplementedError def run(self, message: Message) - Message: try: result self.process(message) result.status done return result except Exception as exc: logger.exception(agent %s processing failed., self.name) message.status failed message.error str(exc) raise AgentError(str(exc)) from exc模型调用统一封装成函数并尽量要求模型返回 JSON 对象减少解析失败概率# aitoearn/core/llm.py import json import re def chat_json(client, model, system_prompt, user_payload, temperature0.5, max_tokens2048): response client.chat.completions.create( modelmodel, temperaturetemperature, max_tokensmax_tokens, messages[ {role: system, content: system_prompt}, {role: user, content: json.dumps(user_payload, ensure_asciiFalse)}, ], response_format{type: json_object}, ) return parse_model_json(response.choices[0].message.content) def parse_model_json(content: str) - dict: try: return json.loads(content) except json.JSONDecodeError: match re.search(r(?:json)?\s*(.*?)\s*, content, re.DOTALL) if match: return json.loads(match.group(1)) raiseparse_model_json先尝试直接解析解析失败再尝试从 Markdown 代码块中提取 JSON。这个兜底逻辑能处理一部分模型输出格式不稳定的情况。需要注意的是response_format参数并非所有模型和兼容端点都支持接入新服务商时先确认能力再启用。4.3 实现策划、创作、审核三个智能体Research 智能体的实现示例# aitoearn/agents/research.py from ..core.agent import BaseAgent from ..core.llm import chat_json from ..core.message import Message SYSTEM_PROMPT 你是一个内容选题策划。 根据用户输入的主题和关键词输出结构化内容简报。 简报必须包含 title_candidates、angle、outline、keywords、target_word_count、platform。 只输出 JSON 对象不要输出多余解释。 class ResearchAgent(BaseAgent): name research def process(self, message: Message) - Message: payload message.payload brief chat_json( self.client, self.model, SYSTEM_PROMPT, payload, temperatureself.temperature, max_tokensself.max_tokens, ) return Message( stepwriter, payload{brief: brief, task_id: payload.get(task_id)}, metadata{source: research, brief_id: brief.get(brief_id)}, )Writer 和 Editor 的结构类似。Editor 的关键在于根据审核结果决定下一步走向# aitoearn/agents/editor.py from ..core.agent import BaseAgent from ..core.llm import chat_json from ..core.message import Message SYSTEM_PROMPT 你是内容审核编辑。 检查稿件是否结构完整、逻辑连贯、关键词合理、内容合规。 输出 passed、score、issues、suggestions 四个字段。 只输出 JSON 对象。 class EditorAgent(BaseAgent): name editor def process(self, message: Message) - Message: payload message.payload review chat_json( self.client, self.model, SYSTEM_PROMPT, payload, temperature0.2, max_tokens1024, ) passed bool(review.get(passed)) return Message( steppublisher if passed else writer, payload{**payload, review: review}, metadata{source: editor, passed: passed}, )step字段在这里充当路由信息审核通过进入publisher不通过回到writer。Editor 的temperature固定为 0.2保证审核结论尽量稳定。4.4 发布智能体与合规边界Publisher 在演示环境先不做真实网络请求而是把发布结果写入本地记录# aitoearn/agents/publisher.py from
