ESSAY
如何利用Agent来做循环检测的工作流
“Agent 自检不是为了完美,而是为了在不完美中找到可靠性。“
核心观点 / 起源
小红书 Worker 上线第一天,验证通过率 0%。
不是 AI 不够聪明,而是它太”自由”了:我要求生成 3-5 个 emoji,它给了我 9 个;我要求恰好 3 个 hashtag,它给了我 10 个;我要求不超过 900 字,它写了 1127 字。
单纯依赖 Prompt 约束无法保证 AI 输出稳定性。即使在 Prompt 中反复强调”必须恰好 3 个 hashtag”,AI 仍然会”创造性地”生成 5-10 个。
如何让 Agent 自己检测和修正输出,而不是每次都依赖人工干预?
Agent 自检循环的核心是建立一个闭环:生成 → 验证 → 评估 → 重试/修正。
┌─────────────────────────────────────────────┐
│ Agent 循环检测工作流 │
└─────────────────────────────────────────────┘
│
┌───────────┴───────────┐
↓ ↓
┌───────────────┐ ┌───────────────┐
│ 生成内容 │ │ 验证规则 │
│ (AI Worker) │ │ (Validation) │
└───────────────┘ └───────────────┘
│ │
└───────────┬───────────┘
↓
┌───────────────┐
│ 自检评估 │
│ (Self-Check) │
└───────────────┘
│
┌───────────┴───────────┐
↓ ↓
通过? 不通过?
│ │
↓ ↓
返回结果 ┌─────────────┐
│ 重试机制 │
│ (Retry) │
└─────────────┘
│
┌───────────┴───────────┐
↓ ↓
重试次数 < 2? 重试次数 ≥ 2?
│ │
↓ ↓
继续重试 ┌─────────────────┐
│ 强制修正 │
│ (Force Fix) │
└─────────────────┘
AI 模型的输出具有随机性,即使使用相同的 Prompt,每次生成的结果也可能不同。单次生成无法保证稳定性,循环检测通过多次尝试和修正,提高输出的可靠性。
细节展开
三层防御的设计逻辑:
- Layer 1(Prompt 约束):预防,尽量让 AI 第一次就生成正确的内容
- Layer 2(智能重试):检测,如果第一次失败,给 AI 第二次机会
- Layer 3(强制修正):兜底,如果多次重试仍失败,用代码强制修正
过程 / 推演
三层防御机制的代码实现
Layer 1: Prompt 约束(预防)
在 Prompt 中强化约束,明确告诉 AI 什么是绝对不能做的。
# xiaohongshu_worker.py - Layer 1: 强化 Prompt 约束
system_prompt = f"""{sop}
【严格禁止】以下表达绝对不能出现:
绝绝子、YYDS、宝子们、家人们、狠狠地、天花板、刺客、真的会谢、建议收藏、干货满满、速速码住、不看后悔、神仙
【严格要求】
- emoji数量必须3-5个(绝对不能超过5个)
- hashtag数量必须恰好3个(不能多也不能少)
- 字数200-900字(绝对不能超过900)
- 不包含任何外链
"""
关键点:使用”绝对不能”、“必须”等强调词,让 AI 理解这是硬性要求而非建议。
Layer 2: 智能重试(检测 + 重试)
生成内容后,立即进行自检验证。如果不符合要求,重新生成,最多重试 2 次。
# xiaohongshu_worker.py - Layer 2: 智能重试机制
max_retries = 2
final_content = None
for attempt in range(max_retries + 1):
# 生成内容
ai_response = requests.post(
f"{base_url}/v1/chat/completions",
headers={"Authorization": f"Bearer {api_key}"},
json={
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"temperature": 0.3
},
timeout=120.0
)
generated_content = ai_response.json()["choices"][0]["message"]["content"]
temp_content = generated_content.replace("[BLOG_URL]", "").replace(blog_url, "")
# 自检:验证格式
emoji_count = len(re.findall(r'[\U0001F300-\U0001F9FF]', temp_content))
hashtag_count = len(re.findall(r'#\w+', temp_content))
# 评估:是否符合要求
if 3 <= emoji_count <= 5 and hashtag_count == 3:
final_content = temp_content
print(f"✅ 第 {attempt + 1} 次尝试成功,格式符合要求")
break
else:
print(f"⚠️ 第 {attempt + 1} 次尝试失败: emoji={emoji_count}, hashtag={hashtag_count}")
# 如果不是最后一次尝试,继续重试
if attempt < max_retries:
continue
关键点:每次生成后立即验证,符合要求就停止,不符合就继续重试。
Layer 3: 强制修正(兜底)
如果重试 2 次仍然失败,使用正则表达式强制修正 emoji 和 hashtag 数量。
# xiaohongshu_worker.py - Layer 3: 强制修正
if attempt == max_retries:
print(f"🔧 达到最大重试次数,启动强制修正")
# 修正 emoji 数量
if emoji_count > 5:
emojis = re.findall(r'[\U0001F300-\U0001F9FF]', temp_content)
for emoji in emojis[5:]:
temp_content = temp_content.replace(emoji, '', 1)
print(f" - Emoji 修正: {emoji_count} → 5")
# 修正 hashtag 数量
hashtags = re.findall(r'#\w+', temp_content)
if len(hashtags) > 3:
for tag in hashtags[3:]:
temp_content = temp_content.replace(tag, '', 1)
print(f" - Hashtag 修正: {len(hashtags)} → 3")
elif len(hashtags) < 3:
generic_tags = ["#干货分享", "#学习笔记", "#技术分享"]
for i in range(3 - len(hashtags)):
temp_content += f" {generic_tags[i]}"
print(f" - Hashtag 补充: {len(hashtags)} → 3")
final_content = temp_content
break
关键点:强制修正是最后的兜底机制,确保无论 AI 生成什么,最终输出都符合要求。
自检循环的关键指标:
| 指标 | 数据 | 说明 |
|---|---|---|
| 平均重试次数 | 1.2 次 | 大部分情况 1-2 次即可通过 |
| 首次成功率 | 45% | 第一次生成即符合要求 |
| 二次成功率 | 85% | 第二次重试后符合要求 |
| 强制修正率 | 15% | 需要强制修正的比例 |
| 最终通过率 | 100% | 三层防御确保最终通过 |
洞察:45% 的首次成功率说明 Prompt 约束有效但不完美,85% 的二次成功率说明智能重试能解决大部分问题,15% 的强制修正率说明兜底机制是必要的。
Agent 结果评估机制
除了格式验证,还需要评估生成内容的质量。使用 AI 评估 AI 的输出,形成质量反馈循环。
def evaluate_content_quality(content: str, platform: str) -> Dict[str, Any]:
"""
使用 AI 评估生成内容的质量
"""
evaluation_prompt = f"""
请评估以下{platform}平台内容的质量:
【评估维度】
1. 调性符合度(1-10分):是否符合{platform}平台风格
2. 内容准确度(1-10分):是否准确传达原文核心观点
3. 可读性(1-10分):是否易于阅读和理解
4. 吸引力(1-10分):是否能吸引用户点击和互动
【内容】
{content}
请以JSON格式返回评分和建议:
{{
"tone_score": 8,
"accuracy_score": 9,
"readability_score": 7,
"engagement_score": 8,
"overall_score": 8.0,
"suggestions": ["建议1", "建议2"]
}}
"""
response = call_ai(evaluation_prompt)
evaluation = json.loads(response)
return evaluation
实际应用:17 篇博客批量处理的质量评估
| 平台 | 平均调性分 | 平均准确度 | 平均可读性 | 平均吸引力 | 综合评分 |
|---|---|---|---|---|---|
| 小红书 | 8.2 | 8.5 | 8.8 | 8.0 | 8.4 |
| X | 9.0 | 9.2 | 9.0 | 9.5 | 9.2 |
| 8.8 | 9.0 | 8.5 | 8.2 | 8.6 | |
| 知乎 | 8.5 | 9.0 | 8.0 | 7.8 | 8.3 |
洞察:X 平台表现最佳(9.2 分),Hook 设计优秀;知乎吸引力相对较低(7.8 分),需优化开头;小红书调性符合度良好(8.2 分),但仍有提升空间。
一点补充
真实报错案例 1:n8n 长文本推理超时
问题现象:Request timeout after 60000ms。n8n HTTP Request 节点固定超时 60 秒,DeepSeek 处理 5000+ 字文章需要 80-120 秒,无法调整超时参数。影响范围:长文章处理失败率 100%,影响 17 篇中的 12 篇(70.6%)。
排查步骤:检查 n8n 日志发现超时错误 → 测试 API 响应时间发现实际需要 87 秒 → 尝试调整 n8n 配置发现不支持自定义超时。
重构方案对比:增加 n8n 超时(n8n 不支持)、分段处理文章(破坏内容完整性)、迁移到 Windmill(完全解决)。最终采用迁移到 Windmill,支持自定义超时 120 秒。
重构效果:
| 指标 | n8n | Windmill | 改进 |
|---|---|---|---|
| 超时限制 | 60 秒(固定) | 120 秒(可调) | +100% |
| 长文章成功率 | 0% | 100% | +100% |
| 平均处理时间 | N/A(超时) | 87 秒 | 可完成 |
真实报错案例 2:小红书验证规则过严导致 0% 通过率
问题现象:emoji=9/5, hashtag=10/3, 验证失败。根本原因:AI 模型输出不稳定,验证规则过严(Emoji 必须 3-5 个,实际生成 6-12 个;Hashtag 必须恰好 3 个,实际生成 5-10 个),缺少后处理修正机制。影响范围:小红书验证通过率 0%(0/17)。
排查步骤:分析批量结果发现平均 emoji 8.2 个、hashtag 7.5 个 → 统计超标分布 → 分析 AI 生成模式发现 AI 倾向于”宁多勿少”。
重构方案:三层防御机制(Prompt 约束 + 智能重试 + 强制修正)。
重构效果:
| 指标 | 优化前 | 优化后 | 改进 |
|---|---|---|---|
| Emoji 合规率 | 0% | 95%+ | +95% |
| Hashtag 合规率 | 0% | 98%+ | +98% |
| 字数合规率 | 70% | 90%+ | +20% |
| 整体通过率 | 0% | 90%+ | +90% |
| 平均重试次数 | N/A | 1.2 次 | 可接受 |
| Token 成本增加 | N/A | +17% | 可接受 |
结语 / 反思
建立 Agent 自检循环的过程,让我对 AI 工程有了新的认识:Agent 自检不是为了完美,而是为了在不完美中找到可靠性。
三层防御机制让 Agent 具备自检和自我修正能力:Prompt 约束预防问题,智能重试检测问题,强制修正兜底问题。从 0% 到 90%+ 的通过率提升,证明了这套机制的有效性。
真实报错案例展示了从问题发现到重构的完整过程:分析根本原因、对比重构方案、验证重构效果。这不仅是技术问题,更是工程思维的体现。
关键是理解 AI 的不确定性,设计合理的评估标准和重试策略,让 Agent 在循环中不断接近目标。