"""提示模板渲染和结构化输出校验。 模板只负责组织输入，Agent rules 和 Runtime 硬规则仍由系统注入，避免 用户模板覆盖知识来源、消费锁和运行状态约束。 """ from __future__ import annotations import json from typing import Any PROMPT_PLACEHOLDERS = { "agent_name", "agent_role", "title", "content_tag", "time_range", "mindmap_node", "product_ref", "next_context", "previous_context", "qc_block", "anchor_schema", } DEFAULT_ROLE_PROMPT = "你是{agent_name}，负责{agent_role}类型的内容任务。" DEFAULT_TASK_TEMPLATE = """请完成以下任务： 标题：{title} 内容标签：{content_tag} 时间范围：{time_range} 导图节点：{mindmap_node} 产品节点：{product_ref} """ DEFAULT_CONSTRAINTS = [ "只围绕本段任务和已提供材料作答，缺少事实时标记待确认。", "输出自然口语，先讲清楚为什么，再给出本段要求的内容。", "正文后输出完整锚点卡 JSON，字段必须与 Schema 一致。", ] DEFAULT_OUTPUT_SCHEMA: dict[str, Any] = { "type": "object", "required": ["core_concepts", "techniques", "products_ref", "golden_lines", "emotion_tone", "ending_hook"], "properties": { "core_concepts": {"type": "string"}, "techniques": {"type": "string"}, "products_ref": {"type": "string"}, "golden_lines": {"type": "string"}, "emotion_tone": {"type": "string"}, "ending_hook": {"type": "string"}, }, } def _string(value: object, limit: int = 20000) -> str: return str(value or "").strip()[:limit] def normalize_template_payload(payload: dict[str, Any], *, partial: bool = False) -> dict[str, Any]: """校验并规范模板输入，供 API 和 seed 共用。""" result: dict[str, Any] = {} if not partial or "name" in payload: name = _string(payload.get("name"), 120) if not name: raise ValueError("提示模板名称不能为空") result["name"] = name for key, limit in (("desc", 500), ("role_prompt", 6000), ("task_template", 12000), ("style", 6000)): if key in payload: result[key] = _string(payload.get(key), limit) if "constraints" in payload: constraints = payload.get("constraints") if not isinstance(constraints, list) or any(not _string(item, 1000) for item in constraints): raise ValueError("提示模板约束必须是非空字符串列表") result["constraints"] = [_string(item, 1000) for item in constraints[:30]] if "examples" in payload: examples = payload.get("examples") if not isinstance(examples, list): raise ValueError("提示模板示例必须是列表") normalized: list[dict[str, str]] = [] for index, item in enumerate(examples[:20], start=1): if not isinstance(item, dict): raise ValueError(f"第 {index} 个提示示例不是对象") example_input = _string(item.get("input"), 4000) example_output = _string(item.get("output"), 6000) if not example_input or not example_output: raise ValueError(f"第 {index} 个提示示例缺少 input 或 output") normalized.append({"input": example_input, "output": example_output}) result["examples"] = normalized if "output_schema" in payload: schema = payload.get("output_schema") if isinstance(schema, str): try: schema = json.loads(schema) except json.JSONDecodeError as exc: raise ValueError("输出 Schema 不是合法 JSON") from exc if not isinstance(schema, dict): raise ValueError("输出 Schema 必须是 JSON 对象") required = schema.get("required", []) if required and (not isinstance(required, list) or any(not isinstance(item, str) for item in required)): raise ValueError("输出 Schema 的 required 必须是字符串列表") result["output_schema"] = schema if "enabled" in payload: result["enabled"] = bool(payload.get("enabled")) return result def template_from_snapshot(snapshot: dict[str, Any] | None) -> dict[str, Any]: data = snapshot if isinstance(snapshot, dict) else {} return { "id": int(data.get("id") or 0), "name": _string(data.get("name") or "默认内容生成模板", 120), "role_prompt": _string(data.get("role_prompt") or DEFAULT_ROLE_PROMPT, 6000), "task_template": _string(data.get("task_template") or DEFAULT_TASK_TEMPLATE, 12000), "constraints": list(data.get("constraints") or DEFAULT_CONSTRAINTS), "examples": list(data.get("examples") or []), "output_schema": dict(data.get("output_schema") or DEFAULT_OUTPUT_SCHEMA), "style": _string(data.get("style") or "大白话、自然口语、适合直接播读", 6000), "version": int(data.get("version") or 1), } def render_prompt(template: dict[str, Any] | None, values: dict[str, object], *, rules: str = "", skills: str = "", memories: str = "", fusion: str = "", anchors: str = "", previous: str = "", knowledge: str = "", runtime_rules: str = "", next_context: str = "", qc_block: str = "", persona: str = "") -> str: """按固定层次拼装提示，模板文本只替换已知变量，保留未知花括号。""" spec = template_from_snapshot(template) def interpolate(text: str) -> str: rendered = text for key in PROMPT_PLACEHOLDERS: rendered = rendered.replace("{" + key + "}", _string(values.get(key))) rendered = rendered.replace("{{" + key + "}}", _string(values.get(key))) return rendered.strip() sections = [ ("Agent 角色", interpolate(spec["role_prompt"])), ("任务模板", interpolate(spec["task_template"])), ] if spec["style"]: sections.append(("模板基础风格（可被 Agent 规则与人格档案覆盖）", interpolate(spec["style"]))) if rules: sections.append(("Agent 规则", rules)) if persona and persona.strip(): sections.append(("人格档案", persona.strip())) if spec["constraints"]: sections.append(("模板约束", "\n".join(f"- {item}" for item in spec["constraints"]))) if spec["examples"]: examples = ["## 示例（仅用于学习格式，不是事实来源）"] for index, item in enumerate(spec["examples"], start=1): examples.append(f"示例 {index} 输入：\n{item.get('input', '')}\n示例 {index} 输出：\n{item.get('output', '')}") sections.append(("上下文示例", "\n\n".join(examples))) for title, content in (("已挂接技能", skills), ("融合协作", fusion), ("已确认记忆", memories), ("上游锚点", anchors), ("上一次任务衔接", previous), ("知识库材料", knowledge), ("运行硬规则", runtime_rules), ("相邻段约束", next_context), ("上次质检意见", qc_block)): if content and content.strip(): sections.append((title, content.strip())) schema = json.dumps(spec["output_schema"], ensure_ascii=False, indent=2) sections.append(("输出协议", f"正文完成后输出锚点卡 JSON，必须符合以下 Schema：\n```json\n{schema}\n```")) return "\n\n".join(f"## {title}\n{content}" for title, content in sections if content) def validate_schema(value: object, schema: dict[str, Any] | None) -> tuple[bool, str]: """提供轻量 JSON Schema 校验，重点覆盖 required/type，避免新增运行依赖。""" if not schema: return True, "" if schema.get("type") == "object" and not isinstance(value, dict): return False, "输出不是 JSON 对象" if isinstance(value, dict): required = schema.get("required") or [] properties = schema.get("properties") or {} missing: list[str] = [] for key in required: if key not in value: missing.append(key) continue rule = properties.get(key) actual = value[key] if isinstance(rule, dict) and rule.get("type") == "string" and not isinstance(actual, str): return False, f"字段 {key} 不是字符串" if actual is None or (isinstance(actual, str) and not actual.strip()): missing.append(key) if missing: return False, "缺少必填字段：" + "、".join(missing) for key, rule in properties.items(): if key not in value or not isinstance(rule, dict) or "type" not in rule: continue expected = rule["type"] actual = value[key] if expected == "string" and not isinstance(actual, str): return False, f"字段 {key} 不是字符串" if expected == "array" and not isinstance(actual, list): return False, f"字段 {key} 不是数组" if expected == "object" and not isinstance(actual, dict): return False, f"字段 {key} 不是对象" return True, ""