恋爱邦
107.39M · 2026-04-09
跑了将近4个月的多 Agent 协作系统,从最初的"能跑就行"到现在对输出质量有了系统化的保障手段,中间踩了不少坑。今天把这套实践整理出来,分享给也在做 AI Agent 工程化的同学。
核心问题很简单:LLM 是概率系统,输出天然不稳定。同一个 prompt,今天输出完美,明天可能就翻车。
那工程上怎么办?
这是最基础的要求。Agent 输出的格式是否符合预期?
# 常见格式问题
# 1. 期望 JSON,返回了 Markdown
# 2. 必填字段缺失
# 3. 数据类型不匹配(期望 number,返回 string)
def validate_output_format(output, schema):
"""验证 Agent 输出格式"""
try:
data = json.loads(output) if isinstance(output, str) else output
except json.JSONDecodeError:
return {"valid": False, "error": "Not valid JSON"}
missing = [f for f in schema["required"] if f not in data]
if missing:
return {"valid": False, "error": f"Missing fields: {missing}"}
return {"valid": True}
解决方案:在 system prompt 中用结构化示例明确格式要求,输出后做 schema 验证,不合格则重试(最多3次)。
格式对了,内容不一定对。
真实案例:Agent 生成一篇技术文章,代码示例的函数名和实际库不一致。格式完美,内容有误。
def verify_content_accuracy(output, context):
"""内容准确性验证"""
checks = []
# 检查代码片段语法
code_blocks = extract_code_blocks(output)
for block in code_blocks:
syntax_ok = check_syntax(block["code"], block["language"])
checks.append({"type": "syntax", "passed": syntax_ok})
# 检查链接是否可访问
links = extract_links(output)
for link in links:
accessible = check_url(link)
checks.append({"type": "link", "passed": accessible})
return checks
内容准确,但是否符合业务规则?
class BusinessValidator:
def __init__(self, rules):
self.rules = rules
def validate(self, content, platform):
results = []
# 检查引流块
if self.rules.get("require_cta"):
has_cta = "Wesley AI 日记" in content
results.append({"rule": "CTA", "passed": has_cta})
# 检查字数限制
max_len = self.rules.get("max_length", {}).get(platform, 50000)
results.append({
"rule": "length",
"passed": len(content) <= max_len
})
# 敏感词检查
sensitive_words = self.rules.get("sensitive_words", [])
found = [w for w in sensitive_words if w in content]
results.append({
"rule": "sensitive",
"passed": len(found) == 0
})
return results
将上面三层验证串联成自动化流水线:
Agent 输出
→ 格式验证(schema check)
→ 内容验证(语法/链接/数据)
→ 业务验证(CTA/敏感词/字数)
→ 全部 PASS → 发布
→ 任一 FAIL → 重试/人工介入
实际运行数据(过去30天):
不要让同一个 Agent 既生成内容又验证质量。
我们增加了一个独立的「质检 Agent」,专门负责审核其他 Agent 的输出。效果非常明显——任务准确率从 60% 提到了 90% 以上。
不是简单地重试,而是带反馈的重试:
def retry_with_feedback(agent, task, max_retries=3):
for attempt in range(max_retries):
output = agent.run(task)
validation = validate(output)
if validation["passed"]:
return output
# 把验证失败的原因反馈给 Agent
task = f"""{task}
上一次输出验证未通过,原因:{validation['errors']}
请修正后重新输出。"""
return None # 升级给人工
测试是发布前的保障,监控是运行时的保障。
# 核心监控指标
METRICS = {
"format_pass_rate": "格式正确率 > 95%",
"content_pass_rate": "内容准确率 > 90%",
"business_pass_rate": "业务合规率 > 95%",
"retry_rate": "重试率 < 20%",
"human_escalation_rate": "人工介入率 < 5%",
}
当任何指标跌破阈值,立即告警,不要等问题累积。
AI Agent 输出质量保障的核心思路:
从「AI 说完成了我就信了」到「AI 说完成了我要验证」,这是 AI Agent 工程化的关键转变。
本文首发于微信公众号「Wesley AI 日记」
AI Agent 实战系列(微信搜索「Wesley AI 日记」关注):
微信搜索「Wesley AI 日记」关注,不错过每一篇更新。