feat(ai-writing): 支持用户粘贴已有大纲并按结构生成

表单可提交 userOutline,后端解析/归一用户大纲并关联素材;补充大纲测试用例与右侧编辑器滑词/段落菜单文档。
This commit is contained in:
fengxiangdebaozi committed 2026-06-03 11:43:44 +08:00
1 parent 3b5525a0a5
commit ab876fc7d8
10 files changed
+944 -3

No files matched your search

@@ -7,11 +7,19 @@ import logging
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.runnables import RunnableConfig
from ..state import AIWritingState, KeyPoint, Outline, OutlineSection
from ..outline_parser import parse_user_outline
from ..prompts.writer_prompts import (
OUTLINE_SYSTEM, OUTLINE_USER, OUTLINE_REVISE_SYSTEM, OUTLINE_REVISE_USER,
OUTLINE_MATERIAL_ASSIGN_SYSTEM,
OUTLINE_MATERIAL_ASSIGN_USER,
OUTLINE_NORMALIZE_SYSTEM,
OUTLINE_NORMALIZE_USER,
OUTLINE_REVISE_SYSTEM,
OUTLINE_REVISE_USER,
OUTLINE_SYSTEM,
OUTLINE_USER,
)
from ._utils import get_writing_model, parse_json, _content_to_text, reasoning_delta, ThinkTagParser, safe_get_stream_writer
from ..state import AIWritingState, KeyPoint, Outline, OutlineSection
from ._utils import ThinkTagParser, _content_to_text, get_writing_model, llm_json, parse_json, reasoning_delta, safe_get_stream_writer
logger = logging.getLogger(__name__)
@@ -80,6 +88,108 @@ def _parse_outline(raw: dict) -> Outline:
return Outline(title=raw.get("title", "未命名文章"), sections=sections)
def _derive_title(state: AIWritingState) -> str:
"""用户大纲没给总标题时,用写作意图兜底一个标题。"""
intent = (state.get("user_intent") or "").strip()
if not intent:
return "未命名文章"
return intent[:30].rstrip() + ("…" if len(intent) > 30 else "")
async def _normalize_user_outline_via_llm(state: AIWritingState, config: RunnableConfig, raw: str) -> Outline:
"""确定性解析失败时,调 LLM 把用户大纲原文归一成三级结构(严格保真、不改写)。
再失败则把每个非空行当作一个章节,保证永远产出一份可用大纲、不丢用户内容。
"""
model_name = (state.get("model_name") or "").strip() or None
line_fallback = {
"title": "",
"sections": [
{"section_title": ln.strip(), "key_points": [], "material_ids": []}
for ln in raw.splitlines() if ln.strip()
] or [{"section_title": "正文", "key_points": [], "material_ids": []}],
}
result = await llm_json(
OUTLINE_NORMALIZE_SYSTEM,
OUTLINE_NORMALIZE_USER.format(raw_outline=raw),
line_fallback,
config,
model_name=model_name,
)
return _parse_outline(result)
async def _assign_materials_to_outline(
state: AIWritingState, config: RunnableConfig, outline: Outline, pkg: dict
) -> Outline:
"""为用户大纲的各章节回填 material_ids(结构锁定:只取编号、按章节序号嫁接)。
LLM 只负责「哪个章节配哪些素材」,标题与层级一律以解析结果为准,杜绝改写。失败则
保持 material_ids 为空(writer_draft 自带「取前 2 条素材」兜底)。
"""
sections = outline.get("sections", [])
materials = pkg.get("materials") or []
if not sections or not materials:
return outline
valid_ids = {m["id"] for m in materials}
sections_list = "\n".join(
f"{i + 1}. {s.get('section_title', '')}" for i, s in enumerate(sections)
)
model_name = (state.get("model_name") or "").strip() or None
result = await llm_json(
OUTLINE_MATERIAL_ASSIGN_SYSTEM,
OUTLINE_MATERIAL_ASSIGN_USER.format(
user_intent=state.get("user_intent", ""),
sections_list=sections_list,
materials_summary=_materials_summary(pkg),
),
{"assignments": []},
config,
model_name=model_name,
)
assignments = result.get("assignments")
if not isinstance(assignments, list):
return outline
for i, section in enumerate(sections):
if i >= len(assignments):
break
ids = assignments[i]
if isinstance(ids, list):
section["material_ids"] = [mid for mid in ids if mid in valid_ids]
return outline
async def _build_outline_from_user(
state: AIWritingState, config: RunnableConfig, pkg: dict, raw: str
) -> dict:
"""用户自带大纲分支:解析用户大纲 → 关联素材 → 直接进确认,不调 LLM 重新规划。"""
writer = safe_get_stream_writer()
writer({"type": "outline_chunk", "text": "正在整理你提供的大纲…\n", "is_thinking": True})
outline = parse_user_outline(raw)
if outline is None:
# 确定性解析拿不准(散文化/格式混乱)→ LLM 兜底归一
outline = await _normalize_user_outline_via_llm(state, config, raw)
if not outline.get("title"):
outline = {**outline, "title": _derive_title(state)}
# 仍走了检索 → 把已检索素材关联到用户章节(结构锁定)
outline = await _assign_materials_to_outline(state, config, outline, pkg)
section_count = len(outline.get("sections", []))
return {
"current_outline": outline,
"status": "awaiting_outline_confirm",
"progress_events": [
{"type": "step_start", "agent_name": "作家", "message": "采用你提供的大纲"},
{"type": "outline_ready", "agent_name": "作家", "message": f"已整理你的大纲,共 {section_count} 个章节"},
],
}
async def writer_outline_node(state: AIWritingState, config: RunnableConfig) -> dict:
pkg = state.get("material_package") or {"keywords": [], "materials": [], "summary": ""}
intervention = state.get("last_user_intervention") or {}
@@ -95,6 +205,13 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) ->
current_outline = state.get("current_outline")
is_revise = bool(feedback) and bool(current_outline)
# 用户自带大纲:首轮(尚无大纲、非重规划)且表单粘了大纲原文 → 走「采用用户大纲」分支,
# 跳过 LLM 重新规划。后续若用户在确认卡里点「重新规划」,current_outline 已存在 → 落到
# 下方 is_revise 编辑模式,按反馈调整用户大纲(符合预期)。
user_outline_raw = (state.get("user_outline_raw") or "").strip()
if user_outline_raw and not current_outline and not is_revise:
return await _build_outline_from_user(state, config, pkg, user_outline_raw)
if is_revise:
system_prompt = OUTLINE_REVISE_SYSTEM
user_prompt = OUTLINE_REVISE_USER.format(
@@ -0,0 +1,220 @@
"""把用户粘贴的大纲文本解析成三级 ``Outline`` 结构(确定性解析,不依赖 LLM)。
用户在写作表单里可能直接粘贴一份已有大纲,要求「完全按我的大纲生成」。本模块负责
把各种常见的大纲文本格式严格、保真地转成后端三级结构:
Outline.sections[] = Level 1 章节(section_title)
OutlineSection.key_points[] = Level 2 小节(point_title)
KeyPoint.sub_points[] = Level 3 子要点
兼容的输入格式:
- Markdown 标题:``#`` / ``##`` / ``###`` / ``####``(首个孤立 ``#`` 视为文章标题)
- 中文编号:``一、`` / ``(一)`` / ``第一章`` / ``第一节``
- 阿拉伯编号:``1.`` / ``1、`` / ``1.1`` / ``1.1.1``
- 缩进 bullet:``-`` / ``*`` / ``•`` / ``·``(用缩进深度表示层级)
- 纯文本平铺:每行一个一级标题(短标题、无句末标点)
设计原则:
- **保真不改写**:只识别层级、原样保留标题文本,绝不增删或改写用户内容。
- **多级混排自适应**:把文本里实际出现的各种「层级标记」收集起来排序,按相对外层
顺序映射到 1/2/3 级——所以不依赖固定的标记搭配,``##``+``-`` 或 ``一、``+缩进
都能正确分层。
- **拿不准就交给上层**:识别不出可靠层级(如整段散文)时返回 ``None``,由调用方走
LLM 兜底归一,避免硬把散文拆成错误结构。
"""
from __future__ import annotations
import re
from .state import KeyPoint, Outline, OutlineSection
# ── 行首标记识别 ──────────────────────────────────────────────────────────────
_CN_NUM = "一二三四五六七八九十百千零〇两"
# Markdown 标题:## 标题
_MD_HEADING_RE = re.compile(r"^(#{1,6})\s+(.*\S)\s*$")
# 第一章 / 第1节 / 第一部分
_CN_CHAPTER_RE = re.compile(rf"^第\s*[{_CN_NUM}\d]+\s*[章节部分篇讲回卷](?:[\s::、..))]\s*|\s+)?(.*)$")
# 一、 / 十二、 / 一. / 一.
_CN_DUN_RE = re.compile(rf"^[{_CN_NUM}]+\s*[、..]\s*(.*)$")
# (一) / (一) / (1) / (1)
_PAREN_RE = re.compile(rf"^[((]\s*[{_CN_NUM}\d]+\s*[))]\s*(.*)$")
# 1.1 / 1.1.1(带至少一个点的多级阿拉伯编号)
_ARABIC_DOTTED_RE = re.compile(r"^(\d+(?:\.\d+)+)\s*[、..))]?\s*(.*)$")
# 1. / 1、 / 1) / 1.(单级阿拉伯编号)
_ARABIC_SINGLE_RE = re.compile(r"^\d+\s*[、..))]\s*(.*)$")
# 带圈数字 ①②③ / 罗马 ⅰ
_CIRCLED_RE = re.compile(r"^[①②③④⑤⑥⑦⑧⑨⑩⑪⑫⑬⑭⑮]\s*(.*)$")
# bullet:- / * / • / ·(要求标记后有空白,避免误吞「*强调*」之类)
_BULLET_RE = re.compile(r"^[-*•·]\s+(.*\S)\s*$")
# 标记分组的「外层度」基准值(越小越外层)。group: 0=标题 1=编号 2=缩进。
# 同组内 order 决定相对层级;不同组之间靠 group 排序(标题 < 编号 < 缩进)。
# 编号顺序遵循中文公文惯例:第一章 > 一、 > (一) > 1. > 1.1 > ①。
_NUM_ORDER = {
"cn_chapter": 0.0, # 第一章
"cn_dun": 1.0, # 一、
"paren": 2.0, # (一)
"arabic_single": 3.0, # 1.
"circled": 8.0, # ①
}
def _indent_width(line: str) -> int:
"""计算行首缩进宽度(tab 记 4 列)。"""
width = 0
for ch in line:
if ch == " ":
width += 1
elif ch == "\t":
width += 4
else:
break
return width
def _classify(stripped: str, indent: int) -> tuple[tuple[int, float], str] | None:
"""识别一行的层级标记,返回 ``((group, order), 纯文本)``;无法识别返回 None。
``(group, order)`` 是该行的「层级键」:把全文出现过的不同键排序后,按位置映射到
1/2/3 级。这样不必预设标记搭配,相对顺序对了层级就对。
"""
if not stripped:
return None
# 1) Markdown 标题(group 0,order = 井号数)
m = _MD_HEADING_RE.match(stripped)
if m:
depth = len(m.group(1))
text = m.group(2).strip()
return (0, float(depth)), text
# 2) 各类编号(group 1)
m = _CN_CHAPTER_RE.match(stripped)
if m and stripped.startswith("第"):
return (1, _NUM_ORDER["cn_chapter"]), (m.group(1) or "").strip()
m = _CN_DUN_RE.match(stripped)
if m:
return (1, _NUM_ORDER["cn_dun"]), m.group(1).strip()
m = _ARABIC_DOTTED_RE.match(stripped)
if m:
depth = m.group(1).count(".") + 1 # 1.1 → 2, 1.1.1 → 3
# 多级阿拉伯:order 随层级递增,落在单级(3.0)与圈号(8.0)之间
return (1, 3.0 + depth), m.group(2).strip()
m = _PAREN_RE.match(stripped)
if m:
return (1, _NUM_ORDER["paren"]), m.group(1).strip()
m = _ARABIC_SINGLE_RE.match(stripped)
if m:
return (1, _NUM_ORDER["arabic_single"]), m.group(1).strip()
m = _CIRCLED_RE.match(stripped)
if m:
return (1, _NUM_ORDER["circled"]), m.group(1).strip()
# 3) bullet:用缩进深度区分层级(group 2)
m = _BULLET_RE.match(stripped)
if m:
return (2, float(indent)), m.group(1).strip()
# 4) 纯文本:用缩进深度区分层级(group 3,最内层基准)
return (3, float(indent)), stripped
def _looks_like_prose(text: str) -> bool:
"""判断一行是否更像散文句子而非大纲标题(长、或带句末标点)。"""
if len(text) > 50:
return True
return text.rstrip()[-1:] in "。!?!?;;"
def parse_user_outline(raw: str) -> Outline | None:
"""把用户粘贴的大纲文本解析成三级 ``Outline``;无法可靠解析时返回 ``None``。
返回 ``None`` 的情形:空白输入、只有一行且无任何标记、或整体更像散文段落——这些交给
上层用 LLM 兜底归一,避免误拆。
"""
if not raw or not raw.strip():
return None
rows: list[tuple[tuple[int, float], str]] = []
for line in raw.splitlines():
if not line.strip():
continue
indent = _indent_width(line)
classified = _classify(line.strip(), indent)
if classified is None:
continue
key, text = classified
if text:
rows.append((key, text))
if not rows:
return None
marked = any(key[0] != 3 for key, _ in rows) # 是否出现过标题/编号/bullet 标记
# 纯文本且无任何标记:可能是「平铺标题列表」也可能是散文。
# 单行、或任一行像句子 → 判为散文,返回 None 交给 LLM。
if not marked:
if len(rows) < 2:
return None
if any(_looks_like_prose(text) for _, text in rows):
return None
# 文章标题:仅在「Markdown 首行是孤立的最浅井号、且存在更深井号」时认定。
title = ""
body = rows
heading_depths = [key[1] for key, _ in rows if key[0] == 0]
if heading_depths:
min_depth = min(heading_depths)
first_key = rows[0][0]
is_first_top_heading = first_key == (0, min_depth)
only_one_at_top = sum(1 for d in heading_depths if d == min_depth) == 1
has_deeper = any(d > min_depth for d in heading_depths)
if is_first_top_heading and only_one_at_top and has_deeper:
title = rows[0][1]
body = rows[1:]
if not body:
return None
# 把 body 里出现的不同「层级键」排序,按位置映射到 1/2/3 级(超过 3 级压到 3)。
distinct_keys = sorted({key for key, _ in body})
level_of: dict[tuple[int, float], int] = {
key: min(idx + 1, 3) for idx, key in enumerate(distinct_keys)
}
sections: list[OutlineSection] = []
cur_section: OutlineSection | None = None
cur_point: KeyPoint | None = None
for key, text in body:
level = level_of[key]
if level <= 1:
cur_section = OutlineSection(section_title=text, key_points=[], material_ids=[])
sections.append(cur_section)
cur_point = None
elif level == 2:
if cur_section is None: # 孤立 L2:提升为章节,避免空标题
cur_section = OutlineSection(section_title=text, key_points=[], material_ids=[])
sections.append(cur_section)
cur_point = None
continue
cur_point = KeyPoint(point_title=text, sub_points=[])
cur_section["key_points"].append(cur_point)
else: # level == 3
if cur_section is None: # 孤立 L3:提升为章节
cur_section = OutlineSection(section_title=text, key_points=[], material_ids=[])
sections.append(cur_section)
cur_point = None
elif cur_point is None: # 章节下尚无小节:提升为小节
cur_point = KeyPoint(point_title=text, sub_points=[])
cur_section["key_points"].append(cur_point)
else:
cur_point["sub_points"].append(text)
if not sections:
return None
return Outline(title=title, sections=sections)
@@ -123,6 +123,60 @@ OUTLINE_REVISE_USER = """写作意图:{user_intent}
请在「当前大纲」基础上落实上述「用户修改要求」,未提到的章节保持原样,输出调整后的完整大纲 JSON。"""
# ── 用户自带大纲:LLM 兜底归一(确定性解析失败时才用)──────────────────────────
OUTLINE_NORMALIZE_SYSTEM = """你是一名大纲格式整理助手。
用户粘贴了一份已有的文章大纲(格式可能不规范),你的任务是把它原样转成结构化 JSON。
【最高原则:保真,绝不改写】
- 严格保留用户写的每一条标题文字,逐字照搬,不要润色、改写、翻译、增加或删除任何条目。
- 你只判断层级(哪条是章节、哪条是小节、哪条是子要点),不创作内容。
- 层级判断依据用户原文里的编号(一、/(一)/1./1.1)、Markdown 井号数、或缩进深度。
【三级结构】
- Level 1 章节 → section_title
- Level 2 小节 → key_points[].point_title
- Level 3 子要点 → key_points[].sub_points[](字符串数组)
- 用户大纲只有一级时,每条都是章节,key_points 留空 []。
- 用户大纲只有两级时,sub_points 留空 []。
【输出要求】
- 只输出一个 JSON 对象,不要解释文字,不要 markdown 代码块,必须用英文双引号。
- material_ids 一律输出 [](此步不关联素材)。
- 结构:{"title": "...", "sections": [{"section_title": "...", "key_points": [{"point_title": "...", "sub_points": ["..."]}], "material_ids": []}]}
- 用户大纲若包含一个总标题,放进 title;没有就留空字符串。"""
OUTLINE_NORMALIZE_USER = """用户粘贴的大纲原文:
{raw_outline}
请逐条保留原文,仅判断层级,输出结构化大纲 JSON。"""
# ── 用户自带大纲:素材关联(结构锁定,只挑 material_ids)──────────────────────────
OUTLINE_MATERIAL_ASSIGN_SYSTEM = """你是一名资料编辑。
给定一份【已确定的文章大纲】(章节顺序固定)和一批检索到的素材,你的任务是为每个章节
挑选最相关的素材编号。
【规则】
- 不要修改大纲的任何标题或结构,你只输出每个章节对应的素材编号列表。
- 严格按大纲给出的章节顺序输出,一个章节对应一个编号数组;该章节没有相关素材就给 []。
- material_id 必须来自给定素材清单里出现过的编号,不要编造。
- 一条素材可以分配给多个章节,也可以一个都不分配。
【输出要求】
- 只输出一个 JSON 对象,不要解释文字,不要 markdown 代码块,必须用英文双引号。
- 结构:{"assignments": [["m001","m002"], [], ["m003"]]},数组长度等于章节数、顺序一致。"""
OUTLINE_MATERIAL_ASSIGN_USER = """写作意图:{user_intent}
文章大纲(章节按顺序编号):
{sections_list}
可用素材:
{materials_summary}
请按章节顺序,为每个章节挑选相关的 material_id,输出 assignments 数组。"""
DRAFT_SECTION_SYSTEM = """你是一名专业内容创作者(作家)。
你的目标:根据大纲和素材,写出一段结构清晰、高质量的文章内容。
你的工作背景:你有十年内容创作经验,文字流畅,善于用事实说话。
@@ -125,6 +125,9 @@ class BlockedSection(TypedDict):
class AIWritingState(TypedDict):
# 写作配置
user_intent: str
# 用户在写作表单粘贴的「已有大纲」原文(可选)。非空时 writer_outline 跳过 LLM 规划,
# 直接解析这份大纲并按用户结构生成,做到「完全按我的大纲写」。留空走正常自动规划。
user_outline_raw: str
article_type: str
target_audience: str
word_count_target: int