188 lines
9.0 KiB
Python
188 lines
9.0 KiB
Python
|
"""提示模板渲染和结构化输出校验。
|
|||
|
|
|
|||
|
|
模板只负责组织输入，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, ""
|