"""Prompt engineering for the three-stage Human Voice Rewrite workflow. Implements PRD v1.1 §44 的核心产品约束 —— 任务链不是 "make it more human"： 理解作者原意 → Semantic Anchor → 识别实际存在的生成式 Pattern → Pattern → Human Rewrite Method → 不完整写作支架 → 全文复检 → 只阻塞真问题。 优先级（PRD §10.1）：顾问确认理解 > 顾问补充约束 > 模型初始理解。 Recheck（PRD §28）：第一轮 High Recall，最终复检 High Precision。 """ from __future__ import annotations import json from evidence import KIND_LABEL, PATTERN_CATEGORY from schemas import AnalyzeResponse, RecheckRequest # -------------------------------------------------------------------------- # PRD §13 V1 AI Pattern Library —— 作为分析 taxonomy 注入，不直接全部展示用户 PATTERN_LIBRARY: list[dict[str, str]] = [ {"id": "P01", "name": "Trait Declaration", "signal": "I became more resilient", "rule": "Trait → Behavior / Mental Reaction"}, {"id": "P02", "name": "Explicit Lesson", "signal": "I learned that...", "rule": "Lesson → Change in Judgment"}, {"id": "P03", "name": "Aphorism", "signal": "X was not Y. X was Z.", "rule": "Aphorism → Personal Observation"}, {"id": "P04", "name": "Parallelism", "signal": "A/B/C；连续相同句式", "rule": "Break Symmetry"}, {"id": "P05", "name": "Metaphor Stack", "signal": "同一个意思连续多个比喻", "rule": "Keep One Motif"}, {"id": "P06", "name": "Over-explanation", "signal": "同一意义多次解释", "rule": "Semantic Compression"}, {"id": "P07", "name": "Perfect Growth", "signal": "failure → resilience → growth", "rule": "Calibrated Change"}, {"id": "P08", "name": "Explicit Transition", "signal": "Slowly / Now I realized / At first", "rule": "Natural Transition"}, {"id": "P09", "name": "Abstract Nouns", "signal": "growth / resilience / uncertainty", "rule": "Abstract → Experience / Verb"}, {"id": "P10", "name": "Over-closure", "signal": "结尾连续多次升华", "rule": "One Ending Idea"}, {"id": "P11", "name": "Synonym Rewrite", "signal": "learned → realized → understood", "rule": "Reframe, Don't Replace"}, {"id": "P12", "name": "Literary Overpackaging", "signal": "大量拟人、抽象意象", "rule": "De-rhetorize"}, ] _TASK_CHAIN = """你的任务链（每一步都不能跳过）： 1. 先理解作者原意，不评价质量； 2. 为全文与每一段建立 Semantic Anchor（必须保留的核心意思）； 3. 识别文章中实际存在的生成式写作 Pattern（不是判断"是不是 AI 写的"）； 4. 把每个 Pattern 映射成明确的人工改写动作（Transformation Rule）； 5. 提供有限度的写作支架（起笔方向），不替顾问完成关键表达； 6. 全文复检只阻塞真正需要返工的问题，给出单一返工目标。""" def _paragraphs_block(paragraphs: list[str]) -> str: return "\n\n".join(f"{text}" for idx, text in enumerate(paragraphs, start=1)) def _constraints_block(global_constraints: list[str], paragraph_constraints: dict[str, list[str]]) -> str: lines: list[str] = [] if global_constraints: lines.append("顾问全局要求：" + "；".join(global_constraints)) if paragraph_constraints: lines.append("顾问分段要求：" + "；".join( f"{pid} -> {'；'.join(v)}" for pid, v in paragraph_constraints.items() if v )) if lines: return "\n".join(lines) return "（无）" # -------------------------------------------------------------------------- # 阶段一：文章理解（PRD §8/§9/§31.1） def build_analyze_prompt( prompt: str, word_limit: int | None, paragraphs: list[str], constraints: list[str], ) -> tuple[str, str]: system = f"""{_TASK_CHAIN} 你是 Human Voice Rewrite 工作流的第一阶段：文章理解。这一轮不输出 AI 味问题，只确认你真正读懂了这篇文章。 要求： - 全文与每段都给出接近自然翻译的中文理解；段落理解要让顾问基本能懂原英文在表达什么，禁止只写"建立冲突""人物成长"这类结构术语。 - semantic_anchor 是"这段必须保留的核心意思"，用于防止后续改写改偏。 - 结合 Essay Prompt 判断文章如何回应题目（prompt 为空时 prompt_alignment 各字段给空字符串，不要编造题目）。 - optional_content_opportunity 只判断"哪些地方如果有真实学生素材会更有人味"；禁止编造新人物、新事件、新时间、新数字、新事实、新情绪。没有素材机会就写空字符串。 - 不做综合质量评分。 - paragraphs 数组必须与输入段落数完全一致：每段一条、按输入顺序；禁止合并、遗漏或拆分段落。 - 只输出 JSON，结构： {{"essay_summary": "", "prompt_alignment": {{"prompt_intent": "", "current_alignment": "", "optional_opportunity": ""}}, "paragraphs": [{{"id": "p1", "original_text": "", "natural_meaning_zh": "", "semantic_anchor": "", "optional_content_opportunity": ""}}]}}""" user = ( f"Essay Prompt：{prompt or '（无，隐藏 prompt 对应分析）'}\n" f"Word Limit：{word_limit if word_limit else '（未提供，不检查）'}\n" f"已有顾问约束：{'；'.join(constraints) if constraints else '（无）'}\n\n" f"文书全文（已按段切分）：\n{_paragraphs_block(paragraphs)}\n\n" f"请输出 JSON。" ) return system, user # -------------------------------------------------------------------------- # 阶段二：Human Voice Diagnosis（PRD §11–§16 / §31.2） def build_diagnose_prompt( prompt: str, word_limit: int | None, paragraphs: list[str], confirmed_anchors: list[str], global_constraints: list[str], paragraph_constraints: dict[str, list[str]], initial_analysis: AnalyzeResponse | None, ) -> tuple[str, str]: library = "\n".join( f"{p['id']} {p['name']}（category={PATTERN_CATEGORY.get(p['id'], 'rhetoric')}；" f"常见表现：{p['signal']}；改写动作：{p['rule']}）" for p in PATTERN_LIBRARY ) categories = "；".join(f"{k}={v}" for k, v in KIND_LABEL.items()) system = f"""{_TASK_CHAIN} 你是 Human Voice Rewrite 工作流的第二阶段：Human Voice Diagnosis。回答"为什么这篇会有 AI 感、哪几段具体有什么问题、人工重写时应该怎么处理"。 输出必须能直接支撑 Workbench 左侧 Focus Mode，而不是只生成一段独立 Coach 文本。 要求： - 输出 Generative Writing Patterns，不输出 AI 概率 / AI Score / 是否 AI 写的判断。 - 只输出当前文章实际命中的 Pattern；没有确凿原文证据不得硬凑。单段默认最多 1–3 个主要 Pattern / Annotation。 - evidence 必须是该段原文中的原句或原短语，逐字引用；禁止改写原文当作证据；禁止输出原文不存在的片段。 - 每个 Pattern 都要说明为什么生成模型容易这么写（why_ai_like）与为什么影响本人感（human_impact）。 - category 只能是 rhetoric / repeat / growth / structure（{categories}）。 - paragraph_briefs 必须与输入段落数完全一致（每段一条、按输入顺序），禁止合并、遗漏或拆分段落，字段含义： - confirmed_meaning：这段在讲什么（以顾问确认后的语义锚点为准，接近自然中文） - rewrite_goal：本轮人工要完成的变化，禁止只写"更自然 / 更具体 / 更像人" - ai_focus：当前段最主要的 1–3 类 AI 味问题，必须能对应原文 evidence；顾问要求保留的表达不得再作为删除目标 - annotations：锚定到原文 Evidence 的批注卡。observation=这里发生了什么；rewrite_action=具体怎么处理（必须可执行）；why_ai_like=为什么这样会显 AI - context_hint：和上下文怎么接。说明上一段已完成什么、当前段只需要完成什么、下一段才该发生什么、哪些 Reflection 不要提前写 - scaffold：给顾问的轻量写作起点（填空 / 思考顺序 / 一句不完整起笔），不得构成完整段落，不得一次给多个成稿选项 - reference_snippet：当前段 1–2 句局部参考，只用于说明改写动作，不得写成完整段落 - 顾问约束优先级最高（PRD §10.1）：confirmed_anchors 非空条目即顾问确认后的语义锚点；顾问要求保留的表达（must_preserve）不得再建议删除。 - optional_suggestion 只给"有真实素材时可以补什么"的可选建议并明确标记可选；没有真实素材时禁止编造，此时给空字符串。 - 不重新规划文章结构、不换故事、不评分。 - 只输出 JSON，结构： {{"overall_diagnosis": "", "patterns": [{{"pattern_id": "P05", "name": "Metaphor Stack", "category": "rhetoric", "affected_paragraphs": ["p1"], "evidence": [""], "why_ai_like": "", "human_impact": "", "transformation_rule": "Keep One Motif"}}], "paragraph_briefs": [{{"paragraph_id": "p1", "confirmed_meaning": "", "rewrite_goal": "", "ai_focus": "", "must_preserve": [], "annotations": [{{"pattern_id": "P05", "category": "rhetoric", "kind_label": "修辞包装", "title": "", "evidence": [""], "observation": "", "rewrite_action": "", "why_ai_like": ""}}], "context_hint": "", "scaffold": "", "reference_snippet": ""}}], "optional_suggestion": ""}}""" anchors = "\n".join( f"{pid}：{anchor}" for pid, anchor in zip([f"p{i}" for i in range(1, len(paragraphs) + 1)], confirmed_anchors) if anchor ) user = ( f"Essay Prompt：{prompt or '（无）'}\n" f"Word Limit：{word_limit if word_limit else '（未提供）'}\n" f"顾问确认后的各段语义锚点（以此为准）：\n{anchors or '（无，以你的理解为默认）'}\n\n" f"顾问约束：\n{_constraints_block(global_constraints, paragraph_constraints)}\n\n" f"第一轮理解摘要（参考）：\n{initial_analysis.essay_summary if initial_analysis else '（无）'}\n\n" f"Pattern 参考库（只输出实际命中的）：\n{library}\n\n" f"文书全文：\n{_paragraphs_block(paragraphs)}\n\n" f"请输出 JSON。" ) return system, user # -------------------------------------------------------------------------- # 阶段三：全文复检（PRD §26–§29 / §31.3） def build_recheck_prompt(req: RecheckRequest) -> tuple[str, str]: diagnosis_block = "" if req.diagnosis: diagnosis_block = json.dumps( { "overall_diagnosis": req.diagnosis.overall_diagnosis, "patterns": [ { "pattern_id": p.pattern_id, "name": p.name, "category": p.category, "affected_paragraphs": p.affected_paragraphs, "evidence": p.evidence, "transformation_rule": p.transformation_rule, } for p in req.diagnosis.patterns ], }, ensure_ascii=False, ) previous_block = "" if req.previous_recheck: previous_block = json.dumps( { "status": req.previous_recheck.status, "checked_rewrite_version": req.previous_recheck.checked_rewrite_version, "revision_targets": [t.model_dump() for t in req.previous_recheck.revision_targets], }, ensure_ascii=False, ) target_block = "" if req.previous_revision_target: target_block = json.dumps(req.previous_revision_target.model_dump(), ensure_ascii=False) system = f"""{_TASK_CHAIN} 你是 Human Voice Rewrite 工作流的最终阶段：全文复检（High Precision —— 只阻塞真正需要返工的问题）。 这是独立 Recheck Prompt，不要复用第一轮 Diagnosis 的教学口吻。 用六个维度检查改后全文（对照原文）： 1. Semantic Preservation：每段是否保持已确认核心意思； 2. Voice Consistency：全文是否像同一个学生在写（语言复杂度、句长、情绪表达、修辞程度、抽象程度、语气）； 3. Original Pattern Reduction：第一轮主要 Pattern 是否真正缓解； 4. New Pattern Emergence：是否出现新的替代模板（例如把 I learned... 换成 I gradually came to realize...，表面词变、底层仍是 Explicit Lesson）； 5. Scaffold / Reference Copying：是否明显高度套用用户实际看过的支架或参考（对照 scaffolds_shown / references_shown）。未实际展开的参考不得作为"用户复制"的依据； 6. Global Coherence：上下文是否自然、是否重复同一 Reflection、是否风格跳变、是否碎片化、{('字数是否超过 ' + str(req.word_limit)) if req.word_limit else '字数（未提供则不检查，word_limit 给 pass）'}。 要求： - 只输出真正 Blocking 的问题；轻微风格差异、个人风格选择一律 pass，不得因此返工（Stop Rule：剩余问题主要属于个人风格选择时给 pass）。 - 返工时只允许 revision_targets 恰一条（首个阻塞段落），每条含单一返工目标 single_revision_goal —— 一句话说明这次只返工一件事。 - 第一轮诊断提过且顾问已合理处理的问题，不得原样重复阻塞，除非它造成了新的严重全文问题并说明冲突。 - 顾问约束优先级最高；顾问要求保留的表达不得因保留本身而被阻塞。 - 若提供了 Previous Recheck / Previous Revision Target：先判断上次 Blocking Issue 是否真正解决，避免反复提出已处理的轻微问题。 - 只输出 JSON，结构： {{"status": "pass", "checked_rewrite_version": "{req.rewrite_version or 'rv_1'}", "global_checks": {{"semantic_preservation": "pass", "voice_consistency": "pass", "pattern_reduction": "pass", "new_pattern": "pass", "coherence": "pass", "reference_copying": "pass", "word_limit": "pass"}}, "revision_targets": [{{"paragraph_id": "p3", "blocking_issue": "", "evidence": [""], "single_revision_goal": ""}}]}} 其中 status 为 "pass" 时 revision_targets 为空数组；global_checks 每项取值 "pass" 或 "fail"。""" paired = "\n\n".join( f"第 {i} 段（{pid}）\n原文：\n{orig}\n\n顾问改写：\n{rew}" for i, (pid, orig, rew) in enumerate( zip( [f"p{i}" for i in range(1, len(req.original_paragraphs) + 1)], req.original_paragraphs, req.rewrite_paragraphs, ), start=1, ) ) anchors = "\n".join( f"{pid}：{anchor}" for pid, anchor in zip([f"p{i}" for i in range(1, len(req.original_paragraphs) + 1)], req.confirmed_anchors) if anchor ) user = ( f"Essay Prompt：{req.prompt or '（无）'}\n" f"Word Limit：{req.word_limit if req.word_limit else '（未提供）'}\n" f"本次 Rewrite Version：{req.rewrite_version or '（未提供）'}\n" f"顾问确认后的语义锚点：\n{anchors or '（无）'}\n\n" f"顾问约束：\n{_constraints_block(req.global_constraints, req.paragraph_constraints)}\n\n" f"第一轮 Diagnosis（用于检查原 Pattern 是否缓解）：\n{diagnosis_block or '（无）'}\n\n" f"顾问实际展开过的写作起点：\n{'；'.join(req.scaffolds_shown) if req.scaffolds_shown else '（无）'}\n\n" f"顾问实际展开过的参考片段：\n{'；'.join(req.references_shown) if req.references_shown else '（无）'}\n\n" f"上一轮 Recheck：\n{previous_block or '（无，这是首次复检）'}\n\n" f"上一轮返工目标：\n{target_block or '（无）'}\n\n" f"全文对照（这是锁定的 Recheck Snapshot，不要对照用户提交后可能继续改动的版本）：\n{paired}\n\n" f"请输出 JSON。" ) return system, user # -------------------------------------------------------------------------- # Progressive Help Level 1：写作起点（PRD §22.2） def build_scaffold_prompt( paragraph_id: str, original_text: str, semantic_anchor: str, rewrite_goal: str, global_constraints: list[str], paragraph_constraints: list[str], ) -> tuple[str, str]: system = f"""{_TASK_CHAIN} 你是 Human Voice Rewrite 的渐进写作支架（Level 1：顾问主动点击「给我一个写作起点」）。 要求： - 只给轻量 Scaffold：填空、思考顺序、一句不完整起笔，或中文问题提示。 - 不构成完整段落；不一次提供多个成稿选项；不自动写成可提交答案。 - 必须贴合语义锚点，服务于当前段 rewrite_goal，并遵守顾问约束。 - 只输出 JSON：{{"scaffold": ""}}""" user = ( f"段落：{paragraph_id}\n" f"原文：\n{original_text}\n\n" f"语义锚点（必须保留的核心意思）：{semantic_anchor}\n" f"本轮改写目标：{rewrite_goal}\n" f"顾问约束：\n{_constraints_block(global_constraints, {paragraph_id: paragraph_constraints} if paragraph_constraints else {})}\n\n" f"请输出 JSON。" ) return system, user # -------------------------------------------------------------------------- # Progressive Help Level 2：参考片段（PRD §22.3） def build_reference_prompt( paragraph_id: str, original_text: str, semantic_anchor: str, rewrite_goal: str, global_constraints: list[str], paragraph_constraints: list[str], ) -> tuple[str, str]: system = f"""{_TASK_CHAIN} 你是 Human Voice Rewrite 的渐进写作支架（Level 2：顾问仍卡住时看一个参考片段）。 要求： - 只给当前段 1–2 句局部参考（starter / reference_snippet）。 - 禁止生成完整当前段；禁止生成完整全文；禁止写成可直接提交的答案。 - 参考只用于说明改写动作，不需要沿用同一句式。 - 必须贴合语义锚点，服务于当前段 rewrite_goal，并遵守顾问约束。 - 只输出 JSON：{{"starter": "", "reference_snippet": ""}} 两个字段写同一段 1–2 句参考即可。""" user = ( f"段落：{paragraph_id}\n" f"原文：\n{original_text}\n\n" f"语义锚点（必须保留的核心意思）：{semantic_anchor}\n" f"本轮改写目标：{rewrite_goal}\n" f"顾问约束：\n{_constraints_block(global_constraints, {paragraph_id: paragraph_constraints} if paragraph_constraints else {})}\n\n" f"请输出 JSON。" ) return system, user def _as_text(value, sep="；") -> str: """模型偶发把字符串字段返回成数组（如 ai_focus 给了 1–3 类问题的列表， 2026-08-21 用户实测 4 段全中）——统一规整为字符串，避免 pydantic 「Input should be a valid string」502；None 归空串（空串是合法默认值）。""" if value is None: return "" if isinstance(value, str): return value if isinstance(value, list): return sep.join(str(x).strip() for x in value if str(x).strip()) return str(value) def _as_list(value): """反向：必须为数组的字段（must_preserve / evidence）模型偶发给成字符串时包一层。""" if isinstance(value, list): return value if isinstance(value, str) and value.strip(): return [value] return [] def normalize_diagnose_payload(data: dict) -> dict: """Tolerate older / loose model fields so Focus Mode still receives a V1.1 shape.""" if not isinstance(data, dict): return data if "overall_diagnosis" in data: data["overall_diagnosis"] = _as_text(data["overall_diagnosis"]) if "optional_suggestion" in data: data["optional_suggestion"] = _as_text(data["optional_suggestion"]) for pat in data.get("patterns") or []: if not isinstance(pat, dict): continue if not pat.get("category"): pat["category"] = PATTERN_CATEGORY.get(str(pat.get("pattern_id") or "").upper(), "rhetoric") for k in ("name", "why_ai_like", "human_impact", "transformation_rule"): if k in pat: pat[k] = _as_text(pat[k]) if "affected_paragraphs" in pat: pat["affected_paragraphs"] = _as_list(pat["affected_paragraphs"]) if "evidence" in pat: pat["evidence"] = _as_list(pat["evidence"]) for brief in data.get("paragraph_briefs") or []: if not isinstance(brief, dict): continue if not brief.get("rewrite_goal") and brief.get("primary_goal"): brief["rewrite_goal"] = brief["primary_goal"] if not brief.get("ai_focus") and brief.get("rewrite_guidance"): brief["ai_focus"] = brief["rewrite_guidance"] if not brief.get("confirmed_meaning") and brief.get("semantic_anchor"): brief["confirmed_meaning"] = brief["semantic_anchor"] if not brief.get("context_hint"): parts = [brief.get("context_before") or "", brief.get("context_after") or ""] brief["context_hint"] = " ".join(p for p in parts if p) if isinstance(brief.get("scaffold"), list): brief["scaffold"] = "\n".join(str(x) for x in brief["scaffold"] if x) if not brief.get("reference_snippet") and brief.get("reference"): brief["reference_snippet"] = brief["reference"] for k in ("confirmed_meaning", "rewrite_goal", "ai_focus", "context_hint", "reference_snippet", "scaffold"): if k in brief: brief[k] = _as_text(brief[k]) if "must_preserve" in brief: brief["must_preserve"] = _as_list(brief["must_preserve"]) notes = brief.get("annotations") if not isinstance(notes, list): brief["annotations"] = [] notes = brief["annotations"] for i, note in enumerate(notes): if not isinstance(note, dict): continue if not note.get("annotation_id"): note["annotation_id"] = f"a{i + 1}" if not note.get("category"): note["category"] = PATTERN_CATEGORY.get(str(note.get("pattern_id") or "").upper(), "rhetoric") if not note.get("kind_label"): note["kind_label"] = KIND_LABEL.get(note["category"], "修辞包装") if not note.get("rewrite_action") and note.get("action"): note["rewrite_action"] = note["action"] for k in ("kind_label", "title", "observation", "rewrite_action", "why_ai_like"): if k in note: note[k] = _as_text(note[k]) if "evidence" in note: note["evidence"] = _as_list(note["evidence"]) return data