Agent Planner输出不是合法JSON怎么办？Schema约束、提取、校验与自动修复
文章摘要
Plan-and-Execute架构依赖Planner生成结构化计划，但模型常出现Markdown代码块、字段缺失、依赖不存在、步骤ID重复或循环依赖。本文给出一套生产级处理链：明确JSON Schema、提取JSON、Pydantic校验、DAG检查、错误反馈修复和失败降级，并提供可直接复用的Python代码。
一、为什么Planner比普通问答更怕格式错误
普通问答多一个标点通常没有影响。
Planner输出却要被程序直接执行：
{
"steps": [
{
"id": "S1",
"task": "查询订单",
"dependencies": []
}
]
}
如果出现以下问题，执行器就可能失败：
- 输出Markdown代码块；
- JSON前后有解释文字；
- 使用单引号；
- 字段名称变化；
steps不是数组；- 步骤ID重复；
- 依赖不存在；
- 循环依赖；
- 工具名称不存在；
- 步骤数量失控。
所以Planner必须经过：
格式解析
→ Schema校验
→ 语义校验
→ 图结构校验
→ 安全校验
二、先定义稳定的数据模型
使用Pydantic 2：
from pydantic import BaseModel, Field
from typing import Literal
class PlanStep(BaseModel):
id: str = Field(pattern=r"^S[1-9][0-9]*$")
title: str = Field(min_length=1, max_length=80)
description: str = Field(min_length=1, max_length=500)
dependencies: list[str] = Field(default_factory=list)
tool_hint: str | None = None
expected_output: str = Field(min_length=1, max_length=300)
risk_level: Literal["low", "medium", "high"] = "low"
class TaskPlan(BaseModel):
objective: str = Field(min_length=1, max_length=500)
assumptions: list[str] = Field(default_factory=list)
steps: list[PlanStep] = Field(min_length=1, max_length=20)
completion_criteria: list[str] = Field(min_length=1)
Schema本身就限制：
- ID格式；
- 字段长度；
- 步骤数量；
- 风险等级枚举；
- 必填字段。
三、Prompt中给出严格输出协议
你是任务规划器。
只输出一个JSON对象，不要输出Markdown代码块、解释、前言或结尾。
输出必须符合以下规则：
1. steps数量为1到20。
2. 步骤ID使用S1、S2、S3格式。
3. dependencies只能引用已经存在的步骤ID。
4. 不允许循环依赖。
5. 每个步骤必须可以独立判断是否完成。
6. 高风险动作标记risk_level=high。
7. 不得生成未注册工具名称。
Prompt可以降低错误率，但不能替代程序校验。
四、提取JSON对象
模型可能返回：
以下是计划：
```json
{...}
可以先去除代码围栏：
```python
import re
def strip_markdown_fence(text: str) -> str:
text = text.strip()
fence = re.fullmatch(
r"```(?:json)?\s*(.*?)\s*```",
text,
flags=re.DOTALL | re.IGNORECASE
)
if fence:
return fence.group(1).strip()
return text
如果前后还有文字，可以提取第一个完整JSON对象。
推荐使用字符状态机，而不是简单正则：
def extract_first_json_object(text: str) -> str:
start = text.find("{")
if start TaskPlan:
clean = strip_markdown_fence(raw_text)
json_text = extract_first_json_object(clean)
try:
data = json.loads(json_text)
except json.JSONDecodeError as exc:
raise ValueError(
f"JSON解析失败：line={exc.lineno}, col={exc.colno}, msg={exc.msg}"
) from exc
try:
return TaskPlan.model_validate(data)
except ValidationError as exc:
raise ValueError(
f"计划Schema校验失败：{exc.errors()}"
) from exc
不要自动把单引号替换成双引号，因为：
- 字符串内部可能包含单引号；
- 错误修复可能改变原意；
- 容易掩盖模型输出质量问题。
六、校验步骤ID与依赖
def validate_dependencies(plan: TaskPlan) -> None:
step_ids = [step.id for step in plan.steps]
if len(step_ids) != len(set(step_ids)):
raise ValueError("存在重复步骤ID")
known = set(step_ids)
for step in plan.steps:
if step.id in step.dependencies:
raise ValueError(
f"步骤{step.id}不能依赖自己"
)
unknown = set(step.dependencies) - known
if unknown:
raise ValueError(
f"步骤{step.id}引用未知依赖：{sorted(unknown)}"
)
还可以要求依赖只能指向前面的步骤：
def validate_dependency_order(plan: TaskPlan) -> None:
positions = {
step.id: index
for index, step in enumerate(plan.steps)
}
for step in plan.steps:
for dependency in step.dependencies:
if positions[dependency] >= positions[step.id]:
raise ValueError(
f"{step.id}依赖了未提前定义的{dependency}"
)
如果执行器支持真正DAG，不强制顺序也可以，但必须做循环检测。
七、检测循环依赖
使用Kahn算法：
from collections import defaultdict, deque
def validate_acyclic(plan: TaskPlan) -> None:
graph: dict[str, list[str]] = defaultdict(list)
indegree: dict[str, int] = {
step.id: 0 for step in plan.steps
}
for step in plan.steps:
for dependency in step.dependencies:
graph[dependency].append(step.id)
indegree[step.id] += 1
queue = deque(
step_id
for step_id, degree in indegree.items()
if degree == 0
)
visited = 0
while queue:
current = queue.popleft()
visited += 1
for next_step in graph[current]:
indegree[next_step] -= 1
if indegree[next_step] == 0:
queue.append(next_step)
if visited != len(plan.steps):
raise ValueError("计划中存在循环依赖")
典型错误：
S1依赖S3
S2依赖S1
S3依赖S2
格式完全合法，但永远无法开始执行。
八、校验工具白名单
REGISTERED_TOOLS = {
"search_web",
"query_order",
"query_inventory",
"generate_report"
}
def validate_tools(plan: TaskPlan) -> None:
for step in plan.steps:
if (
step.tool_hint is not None
and step.tool_hint not in REGISTERED_TOOLS
):
raise ValueError(
f"步骤{step.id}使用未注册工具：{step.tool_hint}"
)
Planner不能凭空创造不存在的工具，执行器必须只接受已注册工具。
九、把错误反馈给模型修复
首次失败后，不要简单重复原Prompt。
应提供明确校验错误：
def build_repair_prompt(
original_request: str,
invalid_output: str,
error_message: str
) -> str:
return f"""
你上一次生成的任务计划无法通过校验。
用户任务：
{original_request}
校验错误：
{error_message}
原始输出：
{invalid_output}
请修复计划。
只输出完整JSON对象，不要输出解释或Markdown代码块。
"""
修复最多尝试1—2次：
def generate_valid_plan(client, user_request: str) -> TaskPlan:
raw = client.complete(build_plan_prompt(user_request))
for attempt in range(2):
try:
plan = parse_plan(raw)
validate_dependencies(plan)
validate_acyclic(plan)
validate_tools(plan)
return plan
except ValueError as exc:
if attempt == 1:
raise
raw = client.complete(
build_repair_prompt(
user_request,
raw,
str(exc)
)
)
raise RuntimeError("无法生成有效计划")
十、不要无限修复
如果连续失败，应降级：
1. 使用更强模型重试一次
2. 切换到简单单步计划
3. 请求用户补充信息
4. 转人工处理
建议记录：
planner_model
attempt_count
validation_error
raw_output_hash
repair_success
十一、结构合法还不够
以下计划Schema正确，但没有执行价值：
{
"steps": [
{
"id": "S1",
"title": "完成任务",
"description": "把任务完成",
"dependencies": [],
"expected_output": "任务完成",
"risk_level": "low"
}
]
}
需要语义规则：
- 步骤必须可执行；
- 不能简单重复用户目标；
- expected_output必须可验证；
- 每一步粒度适中；
- 不得包含未提供的敏感信息；
- 高风险步骤必须标记。
可以增加Reviewer模型，但最终安全边界仍由规则控制。
十二、生产级处理链
模型生成
→ 去除围栏
→ 提取JSON
→ JSON解析
→ Pydantic校验
→ ID与依赖校验
→ DAG循环检测
→ 工具白名单
→ 风险规则
→ 自动修复一次
→ 降级或人工
总结
Planner输出不是合法JSON，不能只靠一句“请严格输出JSON”解决。
可靠方案必须同时具备：
明确Schema
＋解析器
＋类型校验
＋图结构校验
＋工具白名单
＋有限修复
＋失败降级
只有经过校验的计划，才能进入执行器。
