Ascend/gwdoc
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

更新校对错误，修改模型

Browse Source
This commit is contained in:
zhangkunxiang 2026-07-29 09:21:54 +08:00
parent aa4406036e
commit 37da107ade
3 changed files with 542 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

425
app.py
View File

@ -88,6 +88,7 @@ FastAPI 主 Agent 接口 - 修复重复执行问题
"""
import logging
import unicodedata
from difflib import SequenceMatcher
from time import sleep
import aiofiles
@ -121,7 +122,22 @@ from checkpointer_config import (
CheckpointerManager,
checkpointer_manager
)
from gw_write import cotprompt,build_review_prompt,length_convert,length_display,build_length_instruction,dedupe_and_filter,filter_positive_word_false_positives,append_dictionary_results,find_keywords,normalize_words,process_rag_prompt
from gw_write import (
REVIEW_TYPE_ERROR_MAP,
cotprompt,
build_review_prompt,
length_convert,
length_display,
build_length_instruction,
dedupe_and_filter,
filter_positive_word_false_positives,
append_dictionary_results,
find_keywords,
normalize_words,
process_rag_prompt,
get_allowed_review_error_types,
normalize_review_types,
)
app = FastAPI(max_request_size=1024 * 1024 * 10)
_doc2pdf_converter: Optional[Doc2PDF] = None
@ -147,7 +163,8 @@ REWRITE_TYPE_MAP = {
REVIEW_CHUNK_SIZE = 1000
REVIEW_MAX_CONCURRENCY = 5
REVIEW_SYSTEM_PROMPT = (
"你是一个严谨、穷尽式的专业文本校对专家，必须完整检查全文并返回所有符合规则的问题，不能只挑选部分问题。"
"你是一个严谨的专业文本校对专家，必须完整检查用户明确选中的校对范围，并返回该范围内的真实问题。"
"禁止检查或输出用户未选中的校对类别，禁止重复输出相同问题，原文与建议相同时不得输出。"
"校对结果会由程序直接替换进正式公文，因此建议字段只能包含最终正文，任何解释、括号备注或操作说明都会污染公文，绝对禁止输出。"
"每条原文字段必须是输入正文中真实存在的连续原样子串，不能包含任何提前修改后的文字，否则该结果会被系统丢弃。"
"请严格按照用户要求的格式输出。"
@ -311,8 +328,28 @@ WRITE_FORMULA_INSTRUCTION = r"""
7. 若正文不需要公式，不要输出任何 `$` 或 `$$`。
"""
WRITE_HEADING_LAYOUT_INSTRUCTION = """
【标题与正文换行规范（必须遵守）】
1. “一、……”形式的一级标题和“（一）……”形式的二级标题必须分别独占一行，严禁把一级标题、二级标题或正文连续写在同一行。
2. 每个二级标题后必须换行，正文从下一段开始；正文结束后再另起一行输出下一个二级标题。
3. 标题与正文之间、相邻层级标题之间使用一个空行分隔，以便前端 Markdown 正确显示。
4. 必须按以下结构输出，不得写成“（一）标题 正文”：
一、一级标题
（一）二级标题
正文内容。
（二）二级标题
正文内容。
"""
WRITE_SYSTEM_PROMPT = (
"你是专业公文写作助手，请直接输出正文内容。"
"一级标题与二级标题必须分别独占一行，每个二级标题后的正文必须另起一段。"
"输出使用可由前端 Markdown 和 KaTeX 直接解析的内容；如需公式，必须严格遵守用户提示中的公式输出规范。"
)
@ -372,7 +409,12 @@ def build_write_prompt(request: WriteRequest) -> str:
first_level_count,
)
return document_prompt + structure_instruction + WRITE_FORMULA_INSTRUCTION
return (
document_prompt
+ structure_instruction
+ WRITE_HEADING_LAYOUT_INSTRUCTION
+ WRITE_FORMULA_INSTRUCTION
)
async def stream_write_content(prompt: str):
@ -450,6 +492,16 @@ class ReviewRequest(BaseModel):
def check_content(self):
if not self.content or not self.content.strip():
raise ValueError("content不能为空")
normalized_types = normalize_review_types(self.types)
unknown_types = [
review_type for review_type in self.types
if review_type not in REVIEW_TYPE_ERROR_MAP
]
if unknown_types:
raise ValueError(f"不支持的检查类型：{'、'.join(dict.fromkeys(unknown_types))}")
if not normalized_types:
raise ValueError("types不能为空")
self.types = normalized_types
return self
@ -472,6 +524,12 @@ _REVIEW_BLOCK_RE = re.compile(
r"建议[:：]\s*(?P<suggestion>.*?)(?=\n\s*\n|$)",
re.S,
)
_COMPLETED_REVIEW_BLOCK_RE = re.compile(
r"错误[:：]\s*(?P<error>.*?)\s*[\r\n]+"
r"原文[:：]\s*(?P<original>.*?)\s*[\r\n]+"
r"建议[:：]\s*(?P<suggestion>.*?)(?=\r?\n\s*\r?\n|\r?\n?###)",
re.S,
)
def parse_review_output(text: str) -> List[dict]:
text = text.strip()
@ -497,40 +555,302 @@ def parse_review_output(text: str) -> List[dict]:
return result
def _has_repeating_review_tail(text: str) -> bool:
"""检测模型是否开始连续重复同一条完整记录。"""
matches = list(_COMPLETED_REVIEW_BLOCK_RE.finditer(text))
if len(matches) < 2:
return False
def record_key(match) -> tuple:
return (
match.group("error").strip(),
match.group("original").strip(),
match.group("suggestion").strip(),
)
return record_key(matches[-1]) == record_key(matches[-2])
def _split_review_error_types(error: str) -> List[str]:
"""按原顺序拆分并去重校对错误类型。"""
aliases = {
"逻辑错误": "逻辑不通",
"逻辑不顺": "逻辑不通",
"错别字符": "错别字",
}
result = []
for error_type in re.split(r"[、,，]", error or ""):
error_type = error_type.strip()
error_type = aliases.get(error_type.strip(), error_type.strip())
if error_type and error_type not in result:
result.append(error_type)
return result
def _is_whitespace_only_change(original: str, suggestion: str) -> bool:
"""判断原文与建议是否仅存在空白或末尾标点差异。"""
if not original or not suggestion or original == suggestion:
"""判断原文与建议是否相同，或仅存在空白差异。"""
if not original or not suggestion:
return False
def normalize(value: str) -> str:
# 统一全角/半角字符，避免“(4)”与“（4）”被误判为内容修改。
value = unicodedata.normalize("NFKC", str(value))
value = re.sub(r"\s+", "", value)
# 校对模型有时会给原文补充句末冒号、句号等格式标点；
# 若除此之外内容完全一致，则不作为实际错误返回。
return re.sub(r"[，。；：、,.!?！？:;]+$", "", value)
return re.sub(r"\s+", "", value)
return normalize(original) == normalize(suggestion)
_HEADING_PREFIX_RE = re.compile(
r"^\s*(?:"
r"[一二三四五六七八九十百]+、"
r"|[（(][一二三四五六七八九十百\d]+[）)]"
r"|\d+(?:\.\d+)+(?=\s*[\u4e00-\u9fff])"
r"|\d+(?:\.\d+)*[.、]"
r")"
)
_TRAILING_PUNCTUATION_RE = re.compile(r"[，。；：、,.!?！？:;]+$")
_FIGURE_CAPTION_RE = re.compile(
r"^\s*图\s*(?P<number>\d*)\s*[：:]\s*(?P<title>.+?)\s*$"
)
def _is_standalone_heading_punctuation_change(
original: str,
suggestion: str,
) -> bool:
"""过滤只修改独占一行编号标题末尾标点的过度校对。"""
if "\n" in original or "\n" in suggestion:
return False
if not _HEADING_PREFIX_RE.match(original):
return False
original_body = _TRAILING_PUNCTUATION_RE.sub("", original.strip())
suggestion_body = _TRAILING_PUNCTUATION_RE.sub("", suggestion.strip())
return (
original_body == suggestion_body
and original.strip() != suggestion.strip()
)
def _is_figure_numbering_change(
original: str,
suggestion: str,
) -> bool:
"""
分块无法获知全文图号。若标题内容未变，只新增、删除或修改图号，
一律不作为校对结果返回。
"""
original_match = _FIGURE_CAPTION_RE.match(original or "")
suggestion_match = _FIGURE_CAPTION_RE.match(suggestion or "")
if not original_match or not suggestion_match:
return False
return (
original_match.group("title").strip()
== suggestion_match.group("title").strip()
and original_match.group("number")
!= suggestion_match.group("number")
)
def _review_content_skeleton(value: str) -> str:
"""保留字母和数字，用于识别仅标点发生变化的建议。"""
normalized = unicodedata.normalize("NFKC", str(value))
return "".join(
char for char in normalized
if unicodedata.category(char)[:1] in {"L", "N"}
)
def _is_punctuation_only_change(original: str, suggestion: str) -> bool:
return (
bool(original)
and bool(suggestion)
and original != suggestion
and _review_content_skeleton(original)
== _review_content_skeleton(suggestion)
)
def _is_heading_number_change(original: str, suggestion: str) -> bool:
"""识别标题编号变化，避免模型误标为错别字或逻辑错误。"""
original_match = _HEADING_PREFIX_RE.match(original)
suggestion_match = _HEADING_PREFIX_RE.match(suggestion)
if not original_match or not suggestion_match:
return False
return original_match.group() != suggestion_match.group()
def _is_adjacent_repetition_removal(
original: str,
suggestion: str,
) -> bool:
"""识别“常见常见 -> 常见”这类明确的相邻重复删除。"""
if not original or len(original) <= len(suggestion):
return False
removed_length = len(original) - len(suggestion)
if removed_length > 50:
return False
for start in range(0, len(original) - removed_length * 2 + 1):
repeated = original[start:start + removed_length]
if (
repeated
and original[start + removed_length:start + removed_length * 2]
== repeated
and original[:start] + original[start + removed_length:]
== suggestion
):
return True
return False
_CREDENTIAL_EXPOSURE_RE = re.compile(
r"(?i)(?:"
r"(?:postgres(?:ql)?|mysql|mongodb(?:\+srv)?|redis)://[^:\s/]+:[^@\s]+@"
r"|(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*[^\s,，;；]+"
r")"
)
_COMMAND_TEXT_RE = re.compile(
r"(?i)(?:^|\s)(?:--?[a-z][\w-]*|docker(?:\s|$)|dockker(?:\s|$)|doker(?:\s|$))"
)
def _contains_credential_exposure(text: str) -> bool:
return bool(_CREDENTIAL_EXPOSURE_RE.search(text or ""))
def _is_likely_local_text_correction(
original: str,
suggestion: str,
) -> bool:
"""
识别错字、漏字、短语润色和命令拼写等局部修改。
这类修改不能仅通过改写错误标签混入逻辑或合规单选结果。
"""
if not original or not suggestion or original == suggestion:
return False
if _COMMAND_TEXT_RE.search(original):
return True
edits = [
(tag, original[i1:i2], suggestion[j1:j2])
for tag, i1, i2, j1, j2 in SequenceMatcher(
None,
original,
suggestion,
autojunk=False,
).get_opcodes()
if tag != "equal"
]
if not edits:
return False
changed_text = "".join(
original_part + suggestion_part
for _, original_part, suggestion_part in edits
)
if any(char.isdigit() for char in changed_text):
# 年份、日期、数值的前后冲突可能是真正的逻辑问题。
return False
total_changed = sum(
len(original_part) + len(suggestion_part)
for _, original_part, suggestion_part in edits
)
max_span = max(
max(len(original_part), len(suggestion_part))
for _, original_part, suggestion_part in edits
)
return total_changed <= 8 and max_span <= 4
def _minimize_multiline_review_item(
item: dict,
source_text: str,
) -> Optional[dict]:
"""
行式返回协议无法承载多行原文。对于“标签换行值”缺少冒号的情况，
将“名称\\n主发动机 -> 名称：主发动机”缩小为“名称 -> 名称：”。
其他不能安全缩小的多行修改直接丢弃，避免前端只显示第一行后误替换。
"""
original = str(item.get("original", "")).strip()
suggestion = str(item.get("suggestion", "")).strip()
if "\n" not in original and "\r" not in original:
return dict(item)
original_lines = [
line.strip() for line in original.splitlines()
if line.strip()
]
if len(original_lines) < 2 or "\n" in suggestion or "\r" in suggestion:
return None
original_head = original_lines[0]
unchanged_tail = "".join(original_lines[1:])
compact_suggestion = re.sub(r"\s+", "", suggestion)
if not unchanged_tail or not compact_suggestion.endswith(unchanged_tail):
return None
minimized_suggestion = compact_suggestion[:-len(unchanged_tail)]
if (
not minimized_suggestion
or minimized_suggestion == original_head
or original_head not in source_text
):
return None
minimized = dict(item)
minimized["original"] = original_head
minimized["suggestion"] = minimized_suggestion
return minimized
def _normalize_review_item_error(item: dict) -> dict:
"""让明显可判定的错误类型与实际修改保持一致。"""
normalized = dict(item)
original = str(normalized.get("original", "")).strip()
suggestion = str(normalized.get("suggestion", "")).strip()
if _contains_credential_exposure(original):
normalized["error"] = "合规问题"
elif _is_adjacent_repetition_removal(original, suggestion):
normalized["error"] = "重复内容"
elif _is_heading_number_change(original, suggestion):
normalized["error"] = "格式不规范"
elif _is_punctuation_only_change(original, suggestion):
normalized["error"] = "标点不规范"
elif _is_likely_local_text_correction(original, suggestion):
normalized["error"] = "错别词语"
else:
normalized["error"] = "、".join(
_split_review_error_types(normalized.get("error", ""))
)
return normalized
def repair_review_output(
model_output: str,
source_text: str,
positive_words: Optional[Set[str]] = None,
sensitive_words: Optional[Set[str]] = None,
types: Optional[List[str]] = None,
) -> str:
"""修复提前改写原文以及修改范围重叠的校对结果。"""
items = parse_review_output(model_output)
sensitive_word_set = sensitive_words or set()
parsed_items = parse_review_output(model_output)
items = []
for item in parsed_items:
minimized_item = _minimize_multiline_review_item(item, source_text)
if minimized_item is not None:
normalized_item = _normalize_review_item_error(minimized_item)
if (
str(minimized_item.get("original", "")).strip()
in sensitive_word_set
and "敏感词汇" in _split_review_error_types(
minimized_item.get("error", "")
)
):
normalized_item["error"] = "敏感词汇"
items.append(normalized_item)
if not items:
return model_output
return "###\n###"
corrections = [
item for item in items
@ -609,15 +929,59 @@ def repair_review_output(
long_item["error"] = "、".join(long_error_types)
covered_indexes.add(short_index)
final_items = [
item for index, item in enumerate(repaired_items)
if index not in covered_indexes
and item.get("original", "").strip()
and item.get("error", "").strip()
and not _is_whitespace_only_change(
item.get("original", ""), item.get("suggestion", "")
allowed_error_types = (
get_allowed_review_error_types(types)
if types is not None
else None
)
if allowed_error_types is not None and not sensitive_word_set:
allowed_error_types.discard("敏感词汇")
final_items = []
seen_records = set()
for index, item in enumerate(repaired_items):
original = item.get("original", "").strip()
suggestion = item.get("suggestion", "").strip()
error_types = _split_review_error_types(item.get("error", ""))
suggestion_is_valid = bool(suggestion) or "重复内容" in error_types
if (
index in covered_indexes
or not original
or not suggestion_is_valid
or not error_types
or original not in source_text
or (
bool(suggestion)
and _is_whitespace_only_change(original, suggestion)
)
or _is_standalone_heading_punctuation_change(original, suggestion)
or _is_figure_numbering_change(original, suggestion)
or _is_heading_number_change(original, suggestion)
):
continue
if (
allowed_error_types is not None
and any(
error_type not in allowed_error_types
for error_type in error_types
)
):
continue
if (
"敏感词汇" in error_types
and original not in sensitive_word_set
):
continue
normalized_item = dict(item)
normalized_item["error"] = "、".join(error_types)
record_key = (
original,
normalized_item["error"],
suggestion,
)
]
if record_key in seen_records:
continue
seen_records.add(record_key)
final_items.append(normalized_item)
# 正词命中的“错别字”记录不视为错误，例如“维休”被定义为正词时，
# 不返回“维休”→“维修”的错别字建议。
@ -635,6 +999,8 @@ def repair_review_output(
for item in final_items
if item.get("original", "").strip()
]
if not blocks:
return "###\n###"
return "###\n" + "\n\n".join(blocks) + "\n###"
@ -656,12 +1022,16 @@ async def call_review_model(prompt: Iterable[ChatCompletionMessageParam]) -> str
async for chunk in stream_review_model(prompt):
full_text += chunk
if _has_repeating_review_tail(full_text):
logger.warning("[公文校对] 检测到模型连续重复同一条记录，提前结束本块输出")
break
return full_text
def build_review_stream_require(
require: Optional[str],
types: List[str],
sensitive_words: List[str],
negative_words: List[str],
positive_words: List[str],
@ -671,16 +1041,18 @@ def build_review_stream_require(
if require and require.strip():
requirements.append(require.strip())
selected_types = set(normalize_review_types(types))
sensitive_text = "、".join(dict.fromkeys(word.strip() for word in sensitive_words if word and word.strip()))
if sensitive_text:
if sensitive_text and "合规性检查" in selected_types:
requirements.append(f"请重点检查以下敏感词：{sensitive_text}。如果原文出现这些词，错误类型写“敏感词汇”。")
negative_text = "、".join(dict.fromkeys(word.strip() for word in negative_words if word and word.strip()))
if negative_text:
if negative_text and "基础校对" in selected_types:
requirements.append(f"请重点检查以下错误词：{negative_text}。如果原文出现这些词，错误类型写“错误词汇”。")
positive_text = "、".join(dict.fromkeys(word.strip() for word in positive_words if word and word.strip()))
if positive_text:
if positive_text and "基础校对" in selected_types:
requirements.append(f"以下词为正确词，涉及错别字判断时不要误判：{positive_text}。")
if not requirements:
@ -729,6 +1101,7 @@ async def stream_review_content(
):
review_require = build_review_stream_require(
require=require,
types=types,
sensitive_words=sensitive_words,
negative_words=negative_words,
positive_words=positive_words,
@ -754,6 +1127,8 @@ async def stream_review_content(
model_output,
chunk,
positive_words=normalize_words(positive_words),
sensitive_words=normalize_words(sensitive_words),
types=types,
)
# 仅打印过滤后的最终保留记录，不打印模型原始输出或中间结果。
if repaired_output.strip() and repaired_output.strip() != "###\n###":
@ -1796,4 +2171,4 @@ async def root():
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=9090)
uvicorn.run(app, host="0.0.0.0", port=9095)

2
config.py
View File

@ -13,7 +13,7 @@ load_dotenv()
# ==================== 大模型配置 ====================
LLM_CONFIG = {
# "model": "Qwen3.5-35B-A3B",
"model": "46-qwen3.5-35B",
"model": "46-qwen3.6-35B",
# "model": "Qwen3-14B",
"base_url": "http://192.168.0.46:59800/v1",
"api_key": "gpustack_dee9ca823290886c_5edfc86aeeeceb1e9ee5162941cb2cb5",

177
gw_write.py
View File

@ -52,11 +52,41 @@ from openai.types.chat import ChatCompletionSystemMessageParam, ChatCompletionUs
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
REVIEW_TYPE_ERROR_MAP = {
"基础校对": {
"错别字",
"错别词语",
"标点不规范",
"格式不规范",
"重复内容",
"错误词汇",
},
"逻辑校对": {"逻辑不通"},
"合规性检查": {"合规问题", "敏感词汇"},
}
REVIEW_TYPE_ORDER = tuple(REVIEW_TYPE_ERROR_MAP)
def cotprompt(query):
def normalize_review_types(types: List[str]) -> List[str]:
"""按前端传入顺序去重，只保留支持的校对类型。"""
return list(dict.fromkeys(
review_type for review_type in types
if review_type in REVIEW_TYPE_ERROR_MAP
))
def get_allowed_review_error_types(types: List[str]) -> Set[str]:
"""返回本次所选校对项允许输出的错误类型。"""
allowed = set()
for review_type in normalize_review_types(types):
allowed.update(REVIEW_TYPE_ERROR_MAP[review_type])
return allowed
def cotprompt(query, types: Optional[List[str]] = None):
# 精简 Few-Shot：只示范「输入文本 -> 结构化修改结果」，不再包含推理过程/思考链，
# 从源头杜绝模型输出解释、分析、括号备注等多余内容，同时大幅缩短每次请求的前缀长度。
examples = [
basic_examples = [
{
# 综合示例：覆盖错别字/错别词语/标点不规范/重复内容，
# 重点示范：多余标点(。。。)、全半角括号 -> 标点不规范（不是逻辑不通）
@ -74,34 +104,78 @@ def cotprompt(query):
},
{
# 重复内容的原文必须带唯一上下文，建议返回删除后的完整片段。
"question": "2026年通知：2025年通知内容从下述角度开战：一、项目推进方面。各部门需加快项目进度，确保按时交付。\n\n二、培训学习方面：公司将组织系列培训课程，提升员工专业技能。具体时间、地点另行通知。员工需按时参加，不得无辜缺席。自即日起，严禁在工作时间从事与工作无关的活动。\n\n二、纪律要求。近期发现部分员工上班时间做与工作无关的事情，如炒股、聊天等，严重影响工作效率和公司形象。自即日起，严禁在工作时间从事与工作无关的活动。违反规定者，将按照公司规章制度严肃处理。\n\n特此通知：",
"question": "2026年通知：2026年通知内容如下：\n\n一、项目推进方面\n各部门需加快项目进度，确保按时交付。\n\n二、培训学习方面\n公司将组织系列培训课程，提升员工专业技能。具体时间、地点另行通知。员工需按时参加，不得无辜缺席。自即日起，严禁在工作时间从事与工作无关的活动。\n\n二、纪律要求\n近期发现部分员工上班时间做与工作无关的事情，如炒股、聊天等，严重影响工作效率和公司形象。自即日起，严禁在工作时间从事与工作无关的活动。违反规定者，将按照公司规章制度严肃处理。\n\n特此通知：",
"answer": (
"###\n"
"错误：逻辑错误\n"
"原文：2026年通知：2025年通知内容从下述角度开战：\n"
"建议：2026年通知：2026年通知内容从下述角度开展：\n\n"
"错误：标点不规范\n"
"原文：一、项目推进方面。\n"
"建议：一、项目推进方面：\n\n"
"错误：错别词语、重复内容\n"
"原文：培训学习方面：公司将组织系列培训课程，提升员工专业技能。具体时间、地点另行通知。员工需按时参加，不得无辜缺席。自即日起，严禁在工作时间从事与工作无关的活动。\n"
"建议：培训学习方面：公司将组织系列培训课程，提升员工专业技能。具体时间、地点另行通知。员工需按时参加，不得无故缺席。\n\n"
"错误：格式不规范\n"
"原文：二、纪律要求。\n"
"建议：三、纪律要求：\n\n"
"原文：公司将组织系列培训课程，提升员工专业技能。具体时间、地点另行通知。员工需按时参加，不得无辜缺席。自即日起，严禁在工作时间从事与工作无关的活动。\n"
"建议：公司将组织系列培训课程，提升员工专业技能。具体时间、地点另行通知。员工需按时参加，不得无故缺席。\n\n"
"错误：标点不规范\n"
"原文：特此通知：\n"
"建议：特此通知。\n"
"###"
),
},
]
logic_examples = [
{
# 无明显错误示例：只返回空标记，禁止输出任何说明、分析或括号备注
"question": "全年审核采购合同、对账单据数百份，均按规定完成登记归档。",
"question": "项目计划于2026年1月启动，并于2025年12月完成验收。",
"answer": (
"###\n"
"错误：逻辑不通\n"
"原文：项目计划于2026年1月启动，并于2025年12月完成验收。\n"
"建议：项目计划于2026年1月启动，并于2026年12月完成验收。\n"
"###"
),
},
]
logic_scope_examples = [
{
# 单选逻辑校对时，错别字、格式和措辞润色都必须忽略。
"question": "实习培训是理论与时间结合的关键环节。3.9保秘性。第二步：再输入密码连接服务器。",
"answer": "###\n###",
},
]
compliance_examples = [
{
"question": "数据库连接地址为postgresql://admin:password123@10.0.0.8:5432/app。",
"answer": (
"###\n"
"错误：合规问题\n"
"原文：postgresql://admin:password123@10.0.0.8:5432/app\n"
"建议：postgresql://admin:***@10.0.0.8:5432/app\n"
"###"
),
},
]
compliance_scope_examples = [
{
# 普通保密、军用、核心技术等客观表述不因名称本身构成违规。
"question": "本章介绍保密资料、军用设备维修手册、未公开方案和核心技术参数。文中另有“维休”“保秘性”等错字。",
"answer": "###\n###",
},
]
selected_types = normalize_review_types(types or list(REVIEW_TYPE_ORDER))
selected_type_set = set(selected_types)
examples = []
if "基础校对" in selected_type_set:
examples.extend(basic_examples)
if "逻辑校对" in selected_type_set:
examples.extend(logic_examples)
if "合规性检查" in selected_type_set:
examples.extend(compliance_examples)
if selected_type_set == {"逻辑校对"}:
examples.extend(logic_scope_examples)
elif selected_type_set == {"合规性检查"}:
examples.extend(compliance_scope_examples)
examples.append({
"question": "全年审核采购合同、对账单据数百份，均按规定完成登记归档。",
"answer": "###\n###",
})
# 单个示例模板：只保留 问题/答案，删除“推理过程”
example_prompt = PromptTemplate(
input_variables=["question", "answer"],
@ -112,28 +186,29 @@ def cotprompt(query):
examples=examples,
example_prompt=example_prompt,
prefix=(
"你是专业文本校对专家，负责检查错别字、错别词语、错误标点、标题格式与序号、重复内容、逻辑不通及合规问题。\n"
"你是专业文本校对专家，具备检查错别字、错别词语、错误标点、标题格式、重复内容、逻辑不通及合规问题的能力；本次只执行任务要求中明确选中的校对项。\n"
"输出规则（务必严格遵守）：\n"
"1. 每个错误必须连续占三行：第一行“错误：<错误类型>”，第二行“原文：<有错的原文片段>”，第三行“建议：<修改后的正确文本>”。三行之间不得插入空行；不同错误之间空一行。整体用 ### 包裹。\n"
"2. 错误类型只能从以下选择：错别字、错别词语、标点不规范、格式不规范、重复内容、逻辑不通、合规问题。同一连续原文片段存在修改范围重叠的多类错误时，必须合并为一条，错误类型用中文顿号连接，并在一条建议中一次改正。标题序号重复、跳号、顺序错误或层级不一致归为“格式不规范”；标题末尾的纯标点问题归为“标点不规范”。\n"
"2. 错误类型只能从以下选择：错别字、错别词语、标点不规范、格式不规范、重复内容、错误词汇、逻辑不通、合规问题、敏感词汇。同一连续原文片段存在修改范围重叠的多类错误时，必须合并为一条，错误类型用中文顿号连接，并在一条建议中一次改正。\n"
"3. 凡是标点问题（多余或重复标点如“。。。”、全角/半角标点混用如英文括号()应为中文（）、标点缺失或误用），一律归为“标点不规范”，禁止归为“逻辑不通”；“逻辑不通”只用于前后文语义矛盾、指代不清等真正的逻辑问题。\n"
"4. 建议字段只写修改后的正确内容，禁止输出任何解释、分析、推理、评论或括号备注（例如禁止出现“（删除该句……）”“（此处无明显错误……）”“（通常……可接受）”这类内容）。\n"
"4. 建议字段只写修改后的正确内容，禁止输出任何解释、分析、推理、评论或括号备注（例如禁止出现“（删除该句……）”“（此处无明显错误……）”“（通常……可接受）”这类内容）。原文和建议必须确有差异；两者相同或仅空白不同不得输出。相同原文不得重复输出。\n"
"5. 判定重复内容时，必须结合全文语义判断是否造成冗余，不能仅凭文字相同判错。若两处分别承担必要的说明、强调、引用或衔接作用，则不得输出校对项。\n"
"6. 确认属于重复内容后，必须比较每次出现位置与标题主题、前后文和表达作用的匹配程度，保留主题最匹配、逻辑最完整的一处，禁止机械删除第一次或最后一次。\n"
"7. 删除局部重复内容时，原文必须包含足以唯一定位的连续上下文；建议必须返回删除重复内容后的同一完整片段，不得留空。只有整个原文片段均为应删除的纯冗余内容时，建议才可留空。\n"
"8. 必须按错别字和错别词语、标点、标题格式与序号、重复内容、逻辑、合规的顺序逐项检查全文。发现某类错误后仍须继续检查其他类别。\n"
"9. 若整段文本没有任何错误，只返回：###\n###，不要输出其它任何字符。\n"
"8. 独占一行的各级标题允许不带句末标点，不得机械添加冒号或句号。只有原文明确采用“标题与正文在同一行”的格式且标点确有错误时，才校对标题末尾标点。\n"
"9. 原文和建议字段都只能占一行、不得包含换行。若版式把“名称”和“主发动机”等标签与值分成多行，而实际仅缺冒号，应输出最小替换片段，例如“原文：名称”“建议：名称：”，不得把下一行值合并进建议。\n"
"10. 输入是从长文中截取的中间分块，无法获知全文标题序号和图号。禁止新增、删除、替换或重排任何标题序号，不得把“（二）”改为“一、”，也不得把“三、”改为“一、”。“图：故障定位”这类图片说明是允许的原文格式，禁止补成“图1：故障定位”，也禁止新增、删除或修改任何图号。标题序号与图片全局编号均不属于本次校对范围。\n"
"11. 必须仅对本次任务选中的类别逐项检查全文。发现某类错误后仍须继续检查其他已选类别，但禁止检查和输出未选类别。\n"
"12. 若整段文本没有任何错误，只返回：###\n###，不要输出其它任何字符。输出完全部真实问题后立即结束，严禁重复已有记录或为了凑数量继续输出。\n"
"请参照以下示例完成校对："
),
suffix=(
"问题：{question}\n"
"输出答案前必须在内部完成以下扫描，但不要输出扫描过程：\n"
"A. 逐字检查错别字和错别词语；\n"
"B. 逐行检查标题及结尾标点，同级标题格式必须一致；标题后直接接正文时检查是否应使用冒号；\n"
"C. 按正文顺序检查各级标题序号是否重复、跳号、倒序或层级错误；\n"
"D. 全文检索重复句子和连续文本，结合标题主题判断是否冗余；\n"
"E. 继续检查逻辑与合规问题。\n"
"必须完成 A 至 E 后再输出全部问题，不得提前结束或限制问题数量。\n"
"输出答案前只对任务要求中明确选中的类别完成内部扫描，不要输出扫描过程：\n"
"选择基础校对时，检查错别字、错别词语、标点、标题文字格式、重复内容及错误词汇，但不得修改标题序号；\n"
"选择逻辑校对时，检查前后矛盾、时间冲突、因果关系、指代和上下文逻辑；\n"
"选择合规性检查时，检查合规问题及敏感词汇。\n"
"未选中的类别不得检查、修改和输出。输出完真实问题后立即结束，不得重复记录。\n"
"每条原文必须从本次输入正文中逐字、连续、原样复制，严禁提前采用修改结果。建议只能写可直接替换的最终正文。\n"
"答案："
),
@ -147,30 +222,55 @@ def build_review_prompt(types: List[str], content: str, require: Optional[str])
if require is None:
require = ""
require = require.strip()
review_t = "现在需要你帮我完成以下文本校对任务："
for i , t in enumerate(types):
selected_types = normalize_review_types(types)
excluded_types = [
review_type for review_type in REVIEW_TYPE_ORDER
if review_type not in selected_types
]
review_t = "现在需要你帮我完成以下文本校对任务：\n"
for i, t in enumerate(selected_types, start=1):
if t =="逻辑校对":
review_t += f"{i} 对这段文本进行逻辑校对，主要是内容中词语或者其他内容本身是个合规词语，但在文本用词不当，不符合当前内容中逻辑，例如：#2026年终总结：2025年对本公司的业务开战总结，正确的内容应该为：#2026年终总结：2026年对本公司的业务开展总结\n"
review_t += (
f"{i}. 逻辑校对：只检查正文内部有明确证据的前后矛盾、"
"时间先后冲突、因果关系错误、主体或指代冲突。"
"错别字、漏字、标点、编号、命令写法、术语纠正、措辞润色、"
"补充“共几部分”等信息完整性问题均不属于逻辑校对，禁止输出；"
"不能仅凭孤立词句推测逻辑错误。\n"
)
if t =="基础校对":
review_t += f"{i}.1 进行错别字校对，错别字错误主要以文本用字不当为主，例如：星光店电，正确的应该为：星光点点。\n"
review_t += f"{i}.2 进行标点校对，主要以标点符号使用不当为主，例如：星光点点；万里无云，正确的应该为：星光点点，万里无云\n"
review_t += f"{i}.3 进行格式规范校对\n"
review_t += f"{i}.3 进行格式规范校对，但不得修改标题序号或图片编号\n"
review_t += f"{i}.4 进行重复内容校对，主要以原文中出现重复词语、句子或文本为主，例如：“今天今天心情相当不错，我很开心。我很开心。“，正确的内容应该为：今天心情相当不错，我很开心。\n"
if t=="合规性检查":
review_t += f"{i} 对这段文本进行合规性检查\n"
review_t += (
f"{i}. 合规性检查：只检查明文账号口令、访问密钥、令牌等凭据泄露，"
"以及正文中有明确依据的违法违规、隐私泄露或安全合规问题。"
"“保密资料”“军用设备”“未公开方案”“核心技术参数”等客观名称本身不构成错误，"
"不得为了降敏而改写事实。错别字、格式、逻辑和措辞问题均禁止输出。"
"只有额外要求明确提供了敏感词表时，才能输出“敏感词汇”，"
"且原文必须是词表中实际命中的原词。\n"
)
scope_text = (
f"本次选中的校对项：{'、'.join(selected_types)}。\n"
f"本次禁止检查和输出的校对项：{'、'.join(excluded_types) if excluded_types else '无'}。\n"
"只允许输出所选校对项对应的错误类型；未选类别即使发现问题也禁止修改和输出。"
)
require_text = f"\n额外要求：{require}\n" if require else ""
query = f"""
任务要求：
{review_t}
校对范围：
{scope_text}
{require_text}
注意：只输出结构化校对结果（错误/原文/建议），每条结果的错误、原文、建议必须连续三行，三行之间不得有空行；建议只写修改后的正确文本；没有错误时只返回 ###\n###。
必须逐项检查错别字和错别词语、标点、标题格式与序号、重复内容、逻辑及合规。重复内容必须结合标题和上下文判断；删除局部重复内容时，原文必须带唯一上下文，建议必须返回删除后的完整上下文。同一片段中修改范围重叠的多类错误必须合并为一条。
必须仅完整检查上方明确选中的校对项，禁止检查和输出未选类别。重复内容必须结合标题和上下文判断；删除局部重复内容时，原文必须带唯一上下文，建议必须返回删除后的完整上下文。同一片段中修改范围重叠的多类错误必须合并为一条。
待校对正文开始：
{content}
待校对正文结束。
""".strip()
combined_prompt = cotprompt(query)
combined_prompt = cotprompt(query, selected_types)
return [
ChatCompletionUserMessageParam(role="user", content=combined_prompt),
]
@ -274,16 +374,21 @@ def filter_positive_word_false_positives(
for item in items:
original = str(item.get("original", ""))
suggestion = str(item.get("suggestion", ""))
error = str(item.get("error", ""))
has_positive_word = any(word in original for word in positive_words)
changes_positive_word = any(
word in original and word not in suggestion
for word in positive_words
)
is_typo_error = (
"错别字" in error
or "错别词" in error
or "错别词语" in error
)
if has_positive_word and is_typo_error:
if changes_positive_word or (has_positive_word and is_typo_error):
continue
result.append(item)