一、为什么Planner比普通问答更怕格式错误普通问答多一个标点通常没有影响。Planner输出却要被程序直接执行{steps:[{id:S1,task:查询订单,dependencies:[]}]}如果出现以下问题执行器就可能失败输出Markdown代码块JSON前后有解释文字使用单引号字段名称变化steps不是数组步骤ID重复依赖不存在循环依赖工具名称不存在步骤数量失控。所以Planner必须经过格式解析 → Schema校验 → 语义校验 → 图结构校验 → 安全校验二、先定义稳定的数据模型使用Pydantic 2frompydanticimportBaseModel,FieldfromtypingimportLiteralclassPlanStep(BaseModel):id:strField(patternr^S[1-9][0-9]*$)title:strField(min_length1,max_length80)description:strField(min_length1,max_length500)dependencies:list[str]Field(default_factorylist)tool_hint:str|NoneNoneexpected_output:strField(min_length1,max_length300)risk_level:Literal[low,medium,high]lowclassTaskPlan(BaseModel):objective:strField(min_length1,max_length500)assumptions:list[str]Field(default_factorylist)steps:list[PlanStep]Field(min_length1,max_length20)completion_criteria:list[str]Field(min_length1)Schema本身就限制ID格式字段长度步骤数量风险等级枚举必填字段。三、Prompt中给出严格输出协议你是任务规划器。 只输出一个JSON对象不要输出Markdown代码块、解释、前言或结尾。 输出必须符合以下规则 1. steps数量为1到20。 2. 步骤ID使用S1、S2、S3格式。 3. dependencies只能引用已经存在的步骤ID。 4. 不允许循环依赖。 5. 每个步骤必须可以独立判断是否完成。 6. 高风险动作标记risk_levelhigh。 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, flagsre.DOTALL | re.IGNORECASE ) if fence: return fence.group(1).strip() return text如果前后还有文字可以提取第一个完整JSON对象。推荐使用字符状态机而不是简单正则defextract_first_json_object(text:str)-str:starttext.find({)ifstart0:raiseValueError(响应中没有JSON对象)depth0in_stringFalseescapeFalseforindexinrange(start,len(text)):chartext[index]ifescape:escapeFalsecontinueifchar\\andin_string:escapeTruecontinueifchar:in_stringnotin_stringcontinueifin_string:continueifchar{:depth1elifchar}:depth-1ifdepth0:returntext[start:index1]raiseValueError(JSON对象不完整)五、执行JSON和Pydantic校验importjsonfrompydanticimportValidationErrordefparse_plan(raw_text:str)-TaskPlan:cleanstrip_markdown_fence(raw_text)json_textextract_first_json_object(clean)try:datajson.loads(json_text)exceptjson.JSONDecodeErrorasexc:raiseValueError(fJSON解析失败line{exc.lineno}, col{exc.colno}, msg{exc.msg})fromexctry:returnTaskPlan.model_validate(data)exceptValidationErrorasexc:raiseValueError(f计划Schema校验失败{exc.errors()})fromexc不要自动把单引号替换成双引号因为字符串内部可能包含单引号错误修复可能改变原意容易掩盖模型输出质量问题。六、校验步骤ID与依赖defvalidate_dependencies(plan:TaskPlan)-None:step_ids[step.idforstepinplan.steps]iflen(step_ids)!len(set(step_ids)):raiseValueError(存在重复步骤ID)knownset(step_ids)forstepinplan.steps:ifstep.idinstep.dependencies:raiseValueError(f步骤{step.id}不能依赖自己)unknownset(step.dependencies)-knownifunknown:raiseValueError(f步骤{step.id}引用未知依赖{sorted(unknown)})还可以要求依赖只能指向前面的步骤defvalidate_dependency_order(plan:TaskPlan)-None:positions{step.id:indexforindex,stepinenumerate(plan.steps)}forstepinplan.steps:fordependencyinstep.dependencies:ifpositions[dependency]positions[step.id]:raiseValueError(f{step.id}依赖了未提前定义的{dependency})如果执行器支持真正DAG不强制顺序也可以但必须做循环检测。七、检测循环依赖使用Kahn算法fromcollectionsimportdefaultdict,dequedefvalidate_acyclic(plan:TaskPlan)-None:graph:dict[str,list[str]]defaultdict(list)indegree:dict[str,int]{step.id:0forstepinplan.steps}forstepinplan.steps:fordependencyinstep.dependencies:graph[dependency].append(step.id)indegree[step.id]1queuedeque(step_idforstep_id,degreeinindegree.items()ifdegree0)visited0whilequeue:currentqueue.popleft()visited1fornext_stepingraph[current]:indegree[next_step]-1ifindegree[next_step]0:queue.append(next_step)ifvisited!len(plan.steps):raiseValueError(计划中存在循环依赖)典型错误S1依赖S3 S2依赖S1 S3依赖S2格式完全合法但永远无法开始执行。八、校验工具白名单REGISTERED_TOOLS{search_web,query_order,query_inventory,generate_report}defvalidate_tools(plan:TaskPlan)-None:forstepinplan.steps:if(step.tool_hintisnotNoneandstep.tool_hintnotinREGISTERED_TOOLS):raiseValueError(f步骤{step.id}使用未注册工具{step.tool_hint})Planner不能凭空创造不存在的工具执行器必须只接受已注册工具。九、把错误反馈给模型修复首次失败后不要简单重复原Prompt。应提供明确校验错误defbuild_repair_prompt(original_request:str,invalid_output:str,error_message:str)-str:returnf 你上一次生成的任务计划无法通过校验。 用户任务{original_request}校验错误{error_message}原始输出{invalid_output}请修复计划。 只输出完整JSON对象不要输出解释或Markdown代码块。 修复最多尝试1—2次defgenerate_valid_plan(client,user_request:str)-TaskPlan:rawclient.complete(build_plan_prompt(user_request))forattemptinrange(2):try:planparse_plan(raw)validate_dependencies(plan)validate_acyclic(plan)validate_tools(plan)returnplanexceptValueErrorasexc:ifattempt1:raiserawclient.complete(build_repair_prompt(user_request,raw,str(exc)))raiseRuntimeError(无法生成有效计划)十、不要无限修复如果连续失败应降级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 解析器 类型校验 图结构校验 工具白名单 有限修复 失败降级只有经过校验的计划才能进入执行器。
