from typing import Optional, List, Set,Iterable,Any try: import ahocorasick except ImportError: ahocorasick = None import re import json import logging try: from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate except ImportError: try: from langchain.prompts import FewShotPromptTemplate, PromptTemplate except ImportError: class PromptTemplate: def __init__(self, input_variables=None, template: str = ""): self.input_variables = input_variables or [] self.template = template def format(self, **kwargs): return self.template.format(**kwargs) class FewShotPromptTemplate: def __init__( self, examples=None, example_prompt=None, prefix: str = "", suffix: str = "", input_variables=None, example_separator: str = "\n\n", ): self.examples = examples or [] self.example_prompt = example_prompt self.prefix = prefix self.suffix = suffix self.input_variables = input_variables or [] self.example_separator = example_separator def format(self, **kwargs): rendered_examples = [] for example in self.examples: rendered_examples.append(self.example_prompt.format(**example)) parts = [self.prefix] if rendered_examples: parts.append(self.example_separator.join(rendered_examples)) parts.append(self.suffix.format(**kwargs)) return self.example_separator.join(part for part in parts if part) from openai.types.chat import ChatCompletionSystemMessageParam, ChatCompletionUserMessageParam, ChatCompletionMessageParam logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def cotprompt(query): # 精简 Few-Shot：只示范「输入文本 -> 结构化修改结果」，不再包含推理过程/思考链， # 从源头杜绝模型输出解释、分析、括号备注等多余内容，同时大幅缩短每次请求的前缀长度。 examples = [ { # 综合示例：覆盖错别字/错别词语/标点不规范/重复内容， # 重点示范：多余标点(。。。)、全半角括号 -> 标点不规范（不是逻辑不通） "question": "现,代教育中，智能教学系统【】让每个雪生受益，自动文达系统很方便。现就安全生产大检查工作提出如下意见。。。(本报记者 小陈)", "answer": ( "###\n" "错误：标点不规范\n原文：现,代教育\n建议：现代教育\n\n" "错误：标点不规范\n原文：智能教学系统【】\n建议：智能教学系统\n\n" "错误：错别字\n原文：每个雪生\n建议：每个学生\n\n" "错误：错别词语\n原文：自动文达系统\n建议：自动问答系统\n\n" "错误：标点不规范\n原文：如下意见。。。\n建议：如下意见。\n\n" "错误：标点不规范\n原文：(本报记者 小陈)\n建议：（本报记者 小陈）\n" "###" ), }, { # 删除类示例：整段重复/冗余内容应删除时，建议字段留空（表示删除）， # 严禁写“（删除该句……）”之类说明文字，否则前端替换会把说明插入原文。 "question": "综上，各相关部门应认真履行尽责，扎实推进工作落实见效落地。综上，各相关部门应认真履行尽责，扎实推进工作落实见效落地。", "answer": ( "###\n" "错误：重复内容\n原文：综上，各相关部门应认真履行尽责，扎实推进工作落实见效落地。\n建议：\n" "###" ), }, { # 无明显错误示例：只返回空标记，禁止输出任何说明、分析或括号备注 "question": "全年审核采购合同、对账单据数百份，均按规定完成登记归档。", "answer": "###\n###", }, ] # 单个示例模板：只保留 问题/答案，删除“推理过程” example_prompt = PromptTemplate( input_variables=["question", "answer"], template="问题：{question}\n答案：{answer}", ) few_shot_prompt = FewShotPromptTemplate( examples=examples, example_prompt=example_prompt, prefix=( "你是专业文本校对专家，只检查错别字、错误标点、重复内容、逻辑不通与合规问题。\n" "输出规则（务必严格遵守）：\n" "1. 每个错误占三行：第一行“错误：<错误类型>”，第二行“原文：<有错的原文片段>”，第三行“建议：<修改后的正确文本>”；不同错误之间空一行，整体用 ### 包裹。\n" "2. 错误类型只能从以下选择：错别字、错别词语、标点不规范、重复内容、逻辑不通、合规问题。\n" "3. 凡是标点问题（多余或重复标点如“。。。”、全角/半角标点混用如英文括号()应为中文（）、标点缺失或误用），一律归为“标点不规范”，禁止归为“逻辑不通”；“逻辑不通”只用于前后文语义矛盾、指代不清等真正的逻辑问题。\n" "4. 建议字段只写修改后的正确内容，禁止输出任何解释、分析、推理、评论或括号备注（例如禁止出现“（删除该句……）”“（此处无明显错误……）”“（通常……可接受）”这类内容）。\n" "5. 删除类错误（重复插入、冗余整句等需要删掉的内容）：建议字段一律留空，即“建议：”后不写任何字符，用留空表示该原文片段应被删除；绝不能写“（删除）”或删除原因。\n" "6. 若整段文本没有任何错误，只返回：###\n###，不要输出其它任何字符。\n" "请参照以下示例完成校对：" ), suffix="问题：{question}\n答案：", input_variables=["question"], example_separator="\n\n", ) return few_shot_prompt.format(question=query) def build_review_prompt(types: List[str], content: str, require: Optional[str]) -> Iterable[ChatCompletionMessageParam]: if require is None: require = "" require = require.strip() review_t = "现在需要你帮我完成以下文本校对任务：" for i , t in enumerate(types): if t =="逻辑校对": 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}.4 进行重复内容校对，主要以原文中出现重复词语、句子或文本为主，例如：“今天今天心情相当不错，我很开心。我很开心。“，正确的内容应该为：今天心情相当不错，我很开心。\n" if t=="合规性检查": review_t += f"{i} 对这段文本进行合规性检查\n" require_text = f"\n额外要求：{require}\n" if require else "" query = f""" 我会给你对应的文本，进行相应的检查，{review_t} {require_text} 注意：只输出结构化校对结果（错误/原文/建议），建议只写修改后的正确文本，不要输出任何解释、分析、推理或括号备注；没有错误时只返回 ###\n###。 其中：标点问题（多余/重复标点、全半角括号混用等）一律归为“标点不规范”，不要归为“逻辑不通”；需要删除的重复或冗余内容，建议字段留空表示删除，禁止写“（删除…）”之类文字。 """ query1 = cotprompt(content) return [ ChatCompletionUserMessageParam(role="user", content="请不要进行思考，直接输出内容"), ChatCompletionUserMessageParam(role="user", content=query), ChatCompletionUserMessageParam(role="user", content=query1), ] def length_convert(length: Optional[str]): """将篇幅解析为 {target, max}：target 为建议目标字数，max 为硬上限。""" if length == "短": return {"target": 700, "max": 1000} if length == "中": return {"target": 2500, "max": 3000} if length == "长": return {"target": 4500, "max": 5000} # 兜底：从自定义篇幅字符串里解析出最大字数 if length: nums = re.findall(r"\d+", length) if nums: max_chars = int(nums[-1]) return {"target": int(max_chars * 0.8), "max": max_chars} return None def length_display(length: Optional[str]) -> str: """模板 {length} 占位符用的字符串。""" info = length_convert(length) if info: return f"{info['target']}字左右（不超过{info['max']}字）" return length or "" def build_length_instruction(length: Optional[str]) -> str: """把 target/max 写进提示词：建议目标 + 硬上限 + 优先级。""" info = length_convert(length) if not info: return "" return ( f"全文建议写到{info['target']}字左右，绝对不得超过{info['max']}字；" "字数为硬性要求，不得注水或重复凑字数。" ) def normalize_words(words: List[str]) -> Set[str]: return {word.strip() for word in words if word and word.strip()} def find_keywords(text: str, keyword_set: Set[str]) -> List[str]: if not text or not keyword_set: return [] if ahocorasick is None: return [keyword for keyword in keyword_set if keyword in text] automaton = ahocorasick.Automaton() for idx, keyword in enumerate(keyword_set): automaton.add_word(keyword, (idx, keyword)) automaton.make_automaton() found = set() for _, (_, keyword) in automaton.iter(text): found.add(keyword) return list(found) def append_dictionary_results( items: List[dict], content: str, sensitive_words: Set[str], negative_words: Set[str], ) -> List[dict]: for word in find_keywords(content, sensitive_words): items.append({ "original": word, "error": "敏感词汇", "suggestion": "", }) for word in find_keywords(content, negative_words): items.append({ "original": word, "error": "错误词汇", "suggestion": "", }) return items def filter_positive_word_false_positives( items: List[dict], positive_words: Set[str], ) -> List[dict]: if not positive_words: return items result = [] for item in items: original = str(item.get("original", "")) error = str(item.get("error", "")) has_positive_word = any(word in original for word in positive_words) is_typo_error = ( "错别字" in error or "错别词" in error or "错别词语" in error ) if has_positive_word and is_typo_error: continue result.append(item) return result def dedupe_and_filter(items: List[dict], source_text: str) -> List[dict]: seen = set() result = [] for item in items: original = str(item.get("original", "")).strip() error = str(item.get("error", "")).strip() suggestion = str(item.get("suggestion", "")).strip() if not original: continue if original == suggestion: continue if original not in source_text: continue key = (original, error, suggestion) if key in seen: continue seen.add(key) result.append({ "original": original, "error": error, "suggestion": suggestion, }) return result # =============== 辅助函数 =============== def process_rag_prompt(rag_prompt: Any) -> str: """ 处理 rag_prompt 参数，将各种类型转换为字符串 Args: rag_prompt: 可以是字符串、列表、字典等任意类型 Returns: 处理后的字符串 """ if rag_prompt is None: return "" # 如果是字符串，直接返回 if isinstance(rag_prompt, str): return rag_prompt.strip() # 如果是列表，转换为多行字符串 if isinstance(rag_prompt, list): # 将列表中的每个元素转换为字符串，并用换行符连接 return "\n".join(str(item) for item in rag_prompt if item) # 如果是字典，转换为 JSON 字符串 if isinstance(rag_prompt, dict): try: return json.dumps(rag_prompt, ensure_ascii=False, indent=2) except Exception as e: logger.warning(f"rag_prompt 字典转换失败: {e}") return str(rag_prompt) # 其他类型，直接转换为字符串 return str(rag_prompt).strip()