import asyncio import re import sys from typing import Optional, List, Set,Iterable if sys.platform.startswith("win") and hasattr(asyncio, "WindowsSelectorEventLoopPolicy"): asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) import ahocorasick from pydantic import BaseModel, Field, model_validator from fastapi import FastAPI, File, Path, UploadFile, HTTPException, Form, Request, Header, Body, BackgroundTasks from fastapi.responses import StreamingResponse try: from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate except ImportError: from langchain.prompts import FewShotPromptTemplate, PromptTemplate from openai.types.chat import ChatCompletionSystemMessageParam, ChatCompletionUserMessageParam, ChatCompletionMessageParam from chunk_text import get_chunk_bbox, split_text_preserve_sentences, data_replace, chunk_check,merge_short_slices,find_and_read_content_list,process_pdf_file,process_other_file from fastapi.responses import JSONResponse, StreamingResponse import requests from pathlib import PurePath, Path as PathLib import tempfile from modelsAPI import model_api from documents_prompt import ( BEAUTIFY_TYPE_DICT, TEMPLATE_DICT, GENERATE_NORMAL_OUTLINE_PROMPT, GENERATE_MARKDOWN_LIST_OUTLINE_PROMPT, GENERATE_METHOD_OUTLINE_PROMPT, ) """app.py FastAPI 主 Agent 接口 - 修复重复执行问题 提供主脑 Agent 的 HTTP API 服务，支持流式输出 集成 LangGraph Checkpointer 实现状态持久化 """ import logging from time import sleep import aiofiles from dotenv import load_dotenv from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse, FileResponse from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from typing import Optional, Dict, Any, AsyncGenerator,List import uvicorn import json import os import uuid from main_agent import create_main_agent from utils.function_tracker import ( register_event_callback, create_stream_event_handler, unregister_event_callback, clear_current_stream_handler ) import asyncio import importlib from doc2pdf import Doc2PDF from workflow_registry import WORKFLOW_CONFIG, VALID_ROUTE_FLAGS from main_agent import extract_conversation_title from workflows.history_manager import filter_image_urls, preprocess_from_request from config import set_request_user_config from checkpointer_config import ( CheckpointerManager, checkpointer_manager ) app = FastAPI(max_request_size=1024 * 1024 * 10) converter = Doc2PDF() DATA_DIR = "/app/files" load_dotenv() logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # app.mount("/picture", StaticFiles(directory="/app/picture"), name="pictures") STREAM_RESPONSE_HEADERS = { "Cache-Control": "no-cache", "X-Accel-Buffering": "no", "X-Content-Type-Options": "nosniff", } def llm_text_stream_response(content) -> StreamingResponse: return StreamingResponse( content, media_type="text/plain; charset=utf-8", headers=STREAM_RESPONSE_HEADERS, ) @app.on_event("startup") async def startup_event(): """应用启动时初始化""" await checkpointer_manager.setup() # 初始化智能体使用统计表 from tools.agent_usage_statistics import init_agent_usage_table await init_agent_usage_table() logger.info("应用启动完成，PostgreSQL Checkpointer 已配置") # =============== PDF 文件目录配置 =============== # 创建 PDF 文件目录（用于存储生成的 PDF 等文件） PDF_DIR = os.path.join(os.path.dirname(__file__), "created_pdf") # 创建 word 文件目录（用于存储生成的 word 等文件） WORD_DIR = os.path.join(os.path.dirname(__file__), "created_word") os.makedirs(PDF_DIR, exist_ok=True) class RewriteRequest(BaseModel): rewrite_type: Optional[str] = None content: Optional[str] = None require: Optional[str] = None sentence: Optional[str] = None REWRITE_TYPE_MAP = { "quotePolish": "金句润色", "formalStyle": "书面化", "speechStyle": "讲稿化", "formatting": "规整", "expand": "扩写", "continueWriting": "续写", "report": "汇报", "quote": "金句", "summarize": "精简", "condense": "总结", } def build_rewrite_prompt(request: RewriteRequest) -> str: beautify_type = REWRITE_TYPE_MAP.get(request.rewrite_type or "") if not beautify_type: raise HTTPException(status_code=400, detail=f"不支持的 rewrite_type: {request.rewrite_type}") prompt_template = BEAUTIFY_TYPE_DICT.get(beautify_type) if not prompt_template: raise HTTPException(status_code=400, detail=f"未找到提示词模板: {beautify_type}") return prompt_template.format( require=request.require or "", content=request.content or "", sentence=request.sentence or request.content or "", ) async def stream_llm_content(prompt: str): async for chunk in model_api.OpenaiAPI.open_api_chat_stream( query=prompt, model=None, system_prompt="你是专业文稿改写助手，根据要求进行文本改写。", messages=[], ): if chunk: yield chunk @app.post("/Rewrite") async def rewrite(request: RewriteRequest): prompt = build_rewrite_prompt(request) return llm_text_stream_response(stream_llm_content(prompt)) class WriteRequest(BaseModel): role: Optional[str] = None template: Optional[str] = None title: Optional[str] = None length: Optional[str] = None requirement: Optional[str] = None references: Optional[str] = None outline: Optional[str] = None konwlege: Optional[str] = None prompt: Optional[str] = None class OutlineRequest(BaseModel): role: Optional[str] = None template: Optional[str] = None title: Optional[str] = None length: Optional[str] = None requirement: Optional[str] = None references: Optional[str] = None konwlege: Optional[str] = None prompt: Optional[str] = None # aidoc-web 的公文编辑器会把 Markdown 中的 LaTeX 公式交给 KaTeX 渲染。 WRITE_FORMULA_INSTRUCTION = r""" 【公式输出规范（必须遵守）】 1. 仅在正文确实需要数学、统计或技术公式时使用 LaTeX；普通数字、百分比、日期、编号和金额均使用普通文本，不要为了排版而生成公式。 2. 行内公式必须且只能写成 `$公式内容$`，例如：增长率为 $r=\frac{x_2-x_1}{x_1}\times 100\%$。开始和结束的 `$` 之间不得换行。 3. 独占一行的公式必须且只能写成以下形式，起止符各自完整且公式内容位于中间： $$ 公式内容 $$ 4. 只输出 KaTeX 支持的标准 LaTeX。禁止使用 `\(...\)`、`\[...\]`、LaTeX 文档环境、代码块或 ``` 包裹公式；禁止将公式写成图片、HTML、MathML、JSON，也不要转义公式两侧的 `$`。 5. 每个公式的定界符必须成对闭合。公式中的说明文字使用 `\text{...}`；百分号写作 `\%`，乘号优先写作 `\times`。 6. 人民币、预算等金额直接写成“100元”“20万元”等普通文本，不得使用 `$` 作为货币符号，以免被前端误识别为公式。 7. 若正文不需要公式，不要输出任何 `$` 或 `$$`。 """ WRITE_SYSTEM_PROMPT = ( "你是专业公文写作助手，请直接输出正文内容。" "输出使用可由前端 Markdown 和 KaTeX 直接解析的内容；如需公式，必须严格遵守用户提示中的公式输出规范。" ) FIRST_LEVEL_HEADING_PATTERN = re.compile( r"(?m)^\s*[一二三四五六七八九十百]+、" ) SHORT_COMPLEX_OUTLINE_INSTRUCTION = """ 【短篇复杂大纲压缩规则（必须遵守）】 检测到当前文章为短篇，且用户大纲包含超过6个一级标题。为确保全文不超过1000字，必须遵守以下规则： 1. 一级标题最多保留6个。大纲超过6个一级标题时，必须合并内容相近的部分，不得逐项机械展开。 2. 合并大纲时不得删除核心事项，但允许调整标题名称及层级；本规则优先级高于“严格保持原大纲格式、标题不变”的要求。 3. 开头导语控制在120字以内。 4. 每个一级标题的全部内容控制在100字以内。 5. 每个一级标题下最多保留2个二级事项；超过2个时必须合并同类内容。 6. 每个二级事项只使用一句话表述，控制在40字以内，不得继续设置三级标题。 7. 背景、目的和意义应合并表述；纪律要求、成果转化和考核要求应合并表述。 8. 师资介绍、后勤保障、附件说明等次要内容应简写，不得展开评价性、宣传性描述。 9. 参考材料只用于提取时间、地点、对象、内容、要求等关键信息，不得逐段复述。 10. 全文目标约700字，绝对不得超过1000字；篇幅上限优先于大纲层级和参考材料完整复述。 """ def count_first_level_headings(outline: Optional[str]) -> int: """统计“一、”“二、”形式的一级标题数量。""" return len(FIRST_LEVEL_HEADING_PATTERN.findall(outline or "")) 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 build_write_prompt(request: WriteRequest) -> str: prompt = request.prompt or "" template = request.template or "" length_instruction = build_length_instruction(request.length) content = f"请撰写一篇{template}。{length_instruction}{prompt}" prompt_template = TEMPLATE_DICT.get(template) if not prompt_template: raise HTTPException(status_code=400, detail=f"不支持的写作模板: {template}") title = request.title or "" if not title.strip(): raise HTTPException(status_code=400, detail="title不能为空") document_prompt = content + prompt_template.format( role=request.role or "", title=title, length=length_display(request.length), requirement=request.requirement or "", references=request.references or "", outline=request.outline or "", ) structure_instruction = "" first_level_count = count_first_level_headings(request.outline) if request.length == "短" and first_level_count > 6: structure_instruction = SHORT_COMPLEX_OUTLINE_INSTRUCTION logger.info( "[公文撰写] 短篇大纲包含%s个一级标题，启用复杂大纲压缩规则", first_level_count, ) return document_prompt + structure_instruction + WRITE_FORMULA_INSTRUCTION async def stream_write_content(prompt: str): async for chunk in model_api.OpenaiAPI.open_api_chat_stream( query=prompt, model=None, system_prompt=WRITE_SYSTEM_PROMPT, messages=[], ): if chunk: yield chunk @app.post("/Write") async def write(request: WriteRequest): prompt = build_write_prompt(request) return llm_text_stream_response(stream_write_content(prompt)) def build_outline_prompt(request: OutlineRequest) -> str: if request.prompt and request.prompt.strip(): return request.prompt.strip() template = request.template or "" title = request.title or "" if not title.strip(): raise HTTPException(status_code=400, detail="title不能为空") markdown_list_template = {"报告", "函", "文书", "工作总结"} prompt_template = GENERATE_NORMAL_OUTLINE_PROMPT if template == "方案": prompt_template = GENERATE_METHOD_OUTLINE_PROMPT elif template in markdown_list_template: prompt_template = GENERATE_MARKDOWN_LIST_OUTLINE_PROMPT return prompt_template.format( role=request.role or "", template=template, title=title, length=length_display(request.length), requirement=request.requirement or "", references=request.references or "", ) async def stream_outline_content(prompt: str): async for chunk in model_api.OpenaiAPI.open_api_chat_stream( query=prompt, model=None, system_prompt="你是专业公文提纲生成助手，请直接输出提纲内容。", messages=[], ): if chunk: yield chunk @app.post("/Outline") async def outline(request: OutlineRequest): prompt = build_outline_prompt(request) return llm_text_stream_response(stream_outline_content(prompt)) class ReviewRequest(BaseModel): require: Optional[str] = Field(None, description="额外校对要求") content: str = Field(..., description="待校对文本") types: List[str] = Field(..., description="检查类型") sensitive_words: List[str] = Field(default_factory=list, description="敏感词汇") negative_words: List[str] = Field(default_factory=list, description="错误词汇") positive_words: List[str] = Field(default_factory=list, description="正词词汇") @model_validator(mode="after") def check_content(self): if not self.content or not self.content.strip(): raise ValueError("content不能为空") return self class ReviewItem(BaseModel): original: str error: str suggestion: str class ReviewResponse(BaseModel): data: List[ReviewItem] 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 [] 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 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": "一、项目推进方面。各部门需加快项目进度，确保按时交付。\n\n二、培训学习方面：公司将组织系列培训课程，提升员工专业技能。具体时间、地点另行通知。员工需按时参加，不得无辜缺席。自即日起，严禁在工作时间从事与工作无关的活动。\n\n二、纪律要求。近期发现部分员工上班时间做与工作无关的事情，如炒股、聊天等，严重影响工作效率和公司形象。自即日起，严禁在工作时间从事与工作无关的活动。违反规定者，将按照公司规章制度严肃处理。\n\n特此通知：", "answer": ( "###\n" "错误：标点不规范\n" "原文：一、项目推进方面。\n" "建议：一、项目推进方面：\n\n" "错误：错别词语、重复内容\n" "原文：培训学习方面：公司将组织系列培训课程，提升员工专业技能。具体时间、地点另行通知。员工需按时参加，不得无辜缺席。自即日起，严禁在工作时间从事与工作无关的活动。\n" "建议：培训学习方面：公司将组织系列培训课程，提升员工专业技能。具体时间、地点另行通知。员工需按时参加，不得无故缺席。\n\n" "错误：格式不规范\n" "原文：二、纪律要求。\n" "建议：三、纪律要求：\n\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" "7. 重复内容必须采用上下文替换方式：需要删除其中一处时，“原文”不得只写两处都相同的重复句，必须包含待删除位置前后足够且连续的原文，使该片段在全文中能够唯一定位；“建议”必须返回删除重复内容后的同一完整片段，保留其余上下文，不得留空。只有与重复问题无关、且整个原文片段均应删除的纯冗余内容，建议字段才可留空。绝不能写“（删除）”或删除原因。\n" "8. 必须按错别字和错别词语、标点、标题格式与序号、重复内容、逻辑、合规的顺序逐项独立检查全文。发现某一类错误后仍须继续检查其余类别，禁止只输出重复内容而漏掉同段或其他段落中的标点、序号及用词错误；同一连续原文片段存在修改范围重叠的多类错误时，必须合并成一条结果并一次改正。\n" "9. 若整段文本没有任何错误，只返回：###\n###，不要输出其它任何字符。\n" "请参照以下示例完成校对：" ), suffix=( "问题：{question}\n" "输出答案前必须在内部完成以下逐项扫描，但不要输出扫描过程：\n" "A. 逐字检查错别字和错别词语；\n" "B. 逐行检查所有标题及结尾标点，同级标题的标点必须一致；标题后直接接正文时，重点检查是否应使用冒号；\n" "C. 按正文顺序列出所有“一、二、三……”及“（一）（二）（三）……”标题，检查是否重复、跳号或倒序；\n" "D. 在全文中检索重复出现的完整句子和连续文本，结合标题主题判断是否冗余，并删除语义位置不合适的一处；\n" "E. 继续检查逻辑与合规问题。\n" "必须完成 A 至 E 后再输出结果；不得发现部分问题后提前结束，不得限制问题数量，必须输出发现的全部问题。\n" "最终硬性规则：\n" "1. 一级标题后在同一行直接接正文时，标题末尾必须使用冒号；例如“一、项目推进方面。”必须输出为“一、项目推进方面：”，不得漏检。\n" "2. 相同的工作纪律内容同时出现在培训段和纪律段时，纪律段是正确位置；必须保留纪律段，删除培训段中的第一次出现，绝对不得写“保留前一处”。\n" "3. 建议字段会被程序直接替换进正式公文，只能写可直接替换的最终正文；严禁出现“删除该处”“保留前一处”“建议删除”等说明，严禁使用括号解释。\n" "4. 删除局部重复内容时，原文必须带唯一上下文，建议必须是已经删除重复句后的完整上下文，不能只返回重复句。\n" "5. 每条“原文”必须从本次输入正文中逐字、连续、原样复制，必须满足“原文”完整存在于输入正文；严禁在原文字段中提前采用任何修改结果。例如输入是“无辜缺席”，原文必须保留“无辜缺席”，绝不能写成“无故缺席”。\n" "6. 如果带唯一上下文的重复内容片段中同时含有错别字，必须合并为一条“错别词语、重复内容”，原文保留输入中的错误写法，建议一次完成纠词和删除重复句；禁止另外输出一条错别字结果后，再在重复内容的原文字段中使用修改后的文字。\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=query), ChatCompletionUserMessageParam(role="user", content=query1), ] _REVIEW_BLOCK_RE = re.compile( r"错误[:：]\s*(?P.*?)\s*[\r\n]+" r"原文[:：]\s*(?P.*?)\s*[\r\n]+" r"建议[:：]\s*(?P.*?)(?=\n\s*\n|$)", re.S, ) def parse_review_output(text: str) -> List[dict]: text = text.strip() if text.startswith("###"): text = text[3:].strip() if text.endswith("###"): text = text[:-3].strip() result = [] for match in _REVIEW_BLOCK_RE.finditer(text): result.append({ "error": match.group("error").strip(), "original": match.group("original").strip(), "suggestion": match.group("suggestion").strip(), }) return result def _split_review_error_types(error: str) -> List[str]: """按原顺序拆分并去重校对错误类型。""" result = [] for error_type in re.split(r"[、,，]", error or ""): error_type = error_type.strip() if error_type and error_type not in result: result.append(error_type) return result def repair_review_output(model_output: str, source_text: str) -> str: """ 修复模型在“原文”字段中提前使用其他校对建议的问题。 例如正文是“不得无辜缺席”，模型先给出“无辜→无故”，随后又在 重复内容长片段的原文中写成“不得无故缺席”。这里会利用同批次的 独立校对项反向恢复真实原文，并合并被长片段完整覆盖的小校对项。 """ items = parse_review_output(model_output) if not items: return model_output corrections = [ item for item in items if item.get("original") and item.get("suggestion") and item["original"] != item["suggestion"] ] corrections.sort(key=lambda item: len(item["suggestion"]), reverse=True) repaired_items = [] for item in items: repaired = dict(item) # 模型在焊接语境中容易把“核心母的”误判为专业术语“核心母材的”。 # 该错误无法通过普通的原文存在性校验识别，因此在返回前做窄范围纠正。 if ( repaired.get("original", "").strip() == "核心母" and repaired.get("suggestion", "").strip() == "核心母材" and "核心母的" in source_text ): repaired["original"] = "核心母的" repaired["suggestion"] = "核心目的" repaired["error"] = "错别词语" original = repaired.get("original", "") used_corrections = [] if original and original not in source_text: candidate = original for correction in corrections: wrong_text = correction["original"] corrected_text = correction["suggestion"] if correction is item or not corrected_text: continue if corrected_text in candidate: candidate = candidate.replace(corrected_text, wrong_text) used_corrections.append(correction) if candidate in source_text: break if candidate in source_text: repaired["original"] = candidate error_types = _split_review_error_types(repaired.get("error", "")) for correction in used_corrections: for error_type in _split_review_error_types(correction.get("error", "")): if error_type not in error_types: error_types.append(error_type) repaired["error"] = "、".join(error_types) repaired_items.append(repaired) # 如果一个长片段已经同时完成纠词、去重等修改，就不再单独返回被其 # 覆盖的小片段，避免前端先替换小片段后导致长片段无法定位。 covered_indexes = set() for long_index, long_item in enumerate(repaired_items): long_original = long_item.get("original", "") long_suggestion = long_item.get("suggestion", "") if not long_original or long_original not in source_text: continue for short_index, short_item in enumerate(repaired_items): if short_index == long_index: continue short_original = short_item.get("original", "") short_suggestion = short_item.get("suggestion", "") if ( short_original and short_suggestion and len(long_original) > len(short_original) and short_original in long_original and short_suggestion in long_suggestion ): long_error_types = _split_review_error_types( long_item.get("error", "") ) for error_type in _split_review_error_types( short_item.get("error", "") ): if error_type not in long_error_types: long_error_types.append(error_type) 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 ] blocks = [ "\n".join([ f"错误：{item.get('error', '').strip()}", f"原文：{item.get('original', '').strip()}", f"建议：{item.get('suggestion', '').strip()}", ]) for item in final_items if item.get("original", "").strip() ] return "###\n" + "\n\n".join(blocks) + "\n###" REVIEW_SYSTEM_PROMPT = ( "你是一个严谨、穷尽式的专业文本校对专家，必须完整检查全文并返回所有符合规则的问题，不能只挑选部分问题。" "校对结果会由程序直接替换进正式公文，因此建议字段只能包含最终正文，任何解释、括号备注或操作说明都会污染公文，绝对禁止输出。" "每条原文字段必须是输入正文中真实存在的连续原样子串，不能包含任何提前修改后的文字，否则该结果会被系统丢弃。" "请严格按照用户要求的格式输出。" "不要输出推理过程、解释性文字、前缀或总结。" ) async def stream_review_model(prompt: Iterable[ChatCompletionMessageParam]): async for chunk in model_api.OpenaiAPI.open_api_chat_stream( model=None, json_output=False, system_prompt=REVIEW_SYSTEM_PROMPT, messages=prompt, enable_thinking=False, temperature=0.1, ): if chunk: yield chunk async def call_review_model(prompt: Iterable[ChatCompletionMessageParam]) -> str: full_text = "" async for chunk in stream_review_model(prompt): full_text += chunk return full_text def build_review_stream_require( require: Optional[str], sensitive_words: List[str], negative_words: List[str], positive_words: List[str], ) -> Optional[str]: requirements = [] if require and require.strip(): requirements.append(require.strip()) sensitive_text = "、".join(dict.fromkeys(word.strip() for word in sensitive_words if word and word.strip())) if sensitive_text: 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: 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: requirements.append(f"以下词为正确词，涉及错别字判断时不要误判：{positive_text}。") if not requirements: return None return "\n".join(requirements) # /Review 流式并发配置：块大小 5000，最多 5 块并行 REVIEW_CHUNK_SIZE = 5000 REVIEW_MAX_CONCURRENCY = 5 async def stream_review_content( content: str, types: List[str], require: Optional[str], sensitive_words: List[str], negative_words: List[str], positive_words: List[str], ): review_require = build_review_stream_require( require=require, sensitive_words=sensitive_words, negative_words=negative_words, positive_words=positive_words, ) chunks = [ content[i:i + REVIEW_CHUNK_SIZE] for i in range(0, len(content), REVIEW_CHUNK_SIZE) ] if not chunks: return semaphore = asyncio.Semaphore(REVIEW_MAX_CONCURRENCY) async def process_chunk(chunk: str) -> str: async with semaphore: prompt = build_review_prompt( content=chunk, types=types, require=review_require, ) # 累积单块完整输出后再返回：并发块之间不能交错 token， # 否则会破坏“错误/原文/建议”块结构，导致后端解析错乱。 model_output = await call_review_model(prompt) logger.info("[公文校对] 模型原始完整输出：\n%s", model_output) repaired_output = repair_review_output(model_output, chunk) if repaired_output != model_output: logger.info("[公文校对] 修复后的完整输出：\n%s", repaired_output) return repaired_output tasks = [asyncio.create_task(process_chunk(chunk)) for chunk in chunks] # 谁先完成谁先产出（块间顺序不保证；校对条目相互独立，前端按原文定位） for finished in asyncio.as_completed(tasks): chunk_result = await finished if chunk_result and chunk_result.strip(): yield chunk_result yield "\n\n" # 块间分隔，避免相邻块内容被后端按 \n\n 粘连 async def review_chunk( content: str, types: List[str], require: Optional[str], ) -> List[dict]: prompt = build_review_prompt( content=content, types=types, require=require, ) model_output = await call_review_model(prompt) return parse_review_output(model_output) async def review_text( content: str, types: List[str], require: Optional[str], sensitive_words: List[str], negative_words: List[str], positive_words: List[str], ) -> List[dict]: sensitive_word_set = normalize_words(sensitive_words) negative_word_set = normalize_words(negative_words) positive_word_set = normalize_words(positive_words) chunk_size = REVIEW_CHUNK_SIZE chunks = [ content[i:i + chunk_size] for i in range(0, len(content), chunk_size) ] all_items = [] for chunk in chunks: llm_items = await review_chunk( content=chunk, types=types, require=require, ) llm_items = filter_positive_word_false_positives( llm_items, positive_word_set, ) llm_items = append_dictionary_results( llm_items, content=chunk, sensitive_words=sensitive_word_set, negative_words=negative_word_set, ) all_items.extend(llm_items) return dedupe_and_filter(all_items, content) @app.post("/Review") async def review(request: ReviewRequest): return llm_text_stream_response(stream_review_content( content=request.content, types=request.types, require=request.require, sensitive_words=request.sensitive_words, negative_words=request.negative_words, positive_words=request.positive_words, )) # =============== 请求模型 =============== class AgentRequest(BaseModel): """Agent 请求模型""" query: Optional[Any] = None # 用户文本输入 file_text: Optional[str] = None # 文件文本输入 image: Optional[str] = None # 图片路径 audio: Optional[str] = None # 音频路径（支持 audio 或 asr） # 路由标志（可选，如果提供则跳过任务分类） route_flag: Optional[str] = None # 路由标志 history_message: Optional[Any] = None # 历史信息（支持字符串或列表格式） # 接收参数 top_k: Optional[Any] = None # rag 前top_k rag_prompt: Optional[Any] = None # rag 提示词模板（支持字符串、列表、字典等多种类型） image_kb_id: Optional[str] = None # 图库 知识库id image_file_id: Optional[str] = None # 图库 文件id text_kb_id: Optional[str] = None # 检索 知识库id text_file_id: Optional[str] = None # 检索 文件id # Socket.IO 格式所需字段 chat_id: Optional[str] = None # 聊天ID message_id: Optional[str] = None # 消息ID # 知识库配置 x_user_id: Optional[str] = "1" # 用户ID x_user_name: Optional[str] = "testuser" # 用户名称 x_role: Optional[str] = "admin" # 角色 class AgentResponse(BaseModel): """Agent 响应模型""" success: bool message: str data: Optional[Dict[str, Any]] = None error: Optional[str] = None class FeedbackClassifyRequest(BaseModel): """反馈分类请求模型""" feedback_text: str # =============== 辅助函数 =============== 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() def format_stream_event(event_type: str, data: Dict[str, Any], chat_id: str = "", message_id: str = "", type: str = "chat:completion", ) -> str: """格式化流式事件输出为 Socket.IO 格式""" # 生成 message_id（如果没有提供） if not message_id: message_id = str(uuid.uuid4()) # 根据事件类型构建不同的数据结构 if event_type == "stream_content": # stream_content 事件使用 chat:completion 格式 content = data.get("content", "") event_data = { "chat_id": chat_id, "message_id": message_id, "data": { "type": "chat:completion", "data": { "content": content } } } elif event_type == "function_end": # function_end 事件 event_data = { "chat_id": chat_id, "message_id": message_id, "data": { "type": "chat:completion", "data": { "stepsBlueprint": data } } } elif event_type == "execution_complete": # execution_complete 事件 rag_result = data.get("rag_result", {}) graph_result = data.get("graph_result", []) final_result = data.get("final_result", {}) title = data.get("title", "") # 判断是否有有效的 RAG 结果（非空 dict） has_rag = isinstance(rag_result, dict) and bool(rag_result) # 判断是否有有效的图谱结果（非空 list，且至少一个元素的 nodes 非空） has_graph = ( isinstance(graph_result, list) and len(graph_result) > 0 and any(isinstance(item, dict) and item.get("nodes") for item in graph_result) ) # 只有当 RAG 或图谱有有效内容时，才构建 sourceCitation source_citation = None if has_rag or has_graph: source_citation = {} if has_rag: source_citation.update(rag_result) if has_graph: source_citation["xxxx.graph"] = graph_result # 构建 event_data（title 由调用方异步计算后传入） if final_result: content = final_result.get("content", "") event_data = { "chat_id": chat_id, "message_id": message_id, "data": { "type": "chat:completion", "data": { "content": content, }, "actions": final_result.get("actions", ""), "suggestedReplies": final_result.get("suggestedReplies", "") } } else: event_data = { "chat_id": chat_id, "message_id": message_id, "data": { "type": "chat:completion", "data": { "data": {}, "done": True, "title": title or "新对话", } } } # 仅在有引用来源时添加 sourceCitation 字段 if source_citation is not None: event_data["data"]["data"]["sourceCitation"] = source_citation else: # 其他事件 event_data = { "chat_id": chat_id, "message_id": message_id, "type": event_type, "data": { "type": event_type, "data": "" } } # Socket.IO 格式：["chat-events", {...}] socket_io_message = ["chat-events", event_data] return f"{json.dumps(socket_io_message, ensure_ascii=False)}\n" async def execute_workflow_chain( route_flag: str, route_params: Dict[str, Any], chat_id: str = "", message_id: str = "", checkpointer=None ) -> Dict[str, Any]: """ 执行工作流 Args: route_flag: 路由标志 route_params: 路由参数 chat_id: 聊天会话 ID message_id: 消息 ID checkpointer: LangGraph checkpointer 实例 Returns: 工作流执行结果 """ print("子工作流标志：", route_flag) print("子工作流参数：", route_params) if not route_flag: return {} extracted_text = route_params.get("extracted_text", "") combined_query = route_params.get("combined_query", "") history_message = route_params.get("history_message", "") workflow_info = WORKFLOW_CONFIG.get(route_flag) if not workflow_info: logger.warning(f"未知的路由标志: {route_flag}，使用默认的qa工作流") workflow_info = WORKFLOW_CONFIG.get("qa", WORKFLOW_CONFIG.get(list(WORKFLOW_CONFIG.keys())[0])) try: module = importlib.import_module(workflow_info["module"]) workflow_func = getattr(module, workflow_info["function"]) except (ImportError, AttributeError) as e: print(e) logger.error(f"无法导入工作流 {route_flag}: {e}") return {} print("zigongggggggggggggggggggggggg") try: import inspect sig = inspect.signature(workflow_func) params = list(sig.parameters.keys()) kwargs = { "extracted_text": extracted_text, "combined_query": combined_query, "history_message": history_message, "route_flag": route_flag, } if "chat_id" in params: kwargs["chat_id"] = chat_id if "message_id" in params: kwargs["message_id"] = message_id if "checkpointer" in params: kwargs["checkpointer"] = checkpointer result = await workflow_func(**kwargs) except Exception as e: print(e) logger.error(f"工作流执行失败: {e}") return {} return result or {} async def stream_main_agent_execution( raw_input: Dict[str, Any], route_flag: str = "", rag_prompt: str = "", chat_id: str = "", message_id: str = "" ) -> AsyncGenerator[str, None]: """ 流式执行主Agent并输出函数调用信息（使用装饰器事件） Args: raw_input: 原始输入数据 route_flag: 路由标志 rag_prompt: RAG 提示词 chat_id: 聊天会话 ID message_id: 消息 ID Yields: 格式化的流式事件字符串 """ if not chat_id: chat_id = str(uuid.uuid4()) if not message_id: message_id = str(uuid.uuid4()) stream_handler, event_queue = create_stream_event_handler() register_event_callback(stream_handler) try: execution_done = asyncio.Event() execution_error = None final_result = None checkpointer = None async def run_agent(): nonlocal execution_error, final_result, checkpointer route_flag_value: str = "" try: from checkpointer_config import checkpointer_manager checkpointer = await checkpointer_manager.get_async_checkpointer() main_agent = create_main_agent() initial_route_flag: str = route_flag if (route_flag and route_flag in VALID_ROUTE_FLAGS) else "" initial_state = { "raw_input": raw_input, "input_type": "", "has_query":False, "has_image": False, "has_audio": False, "has_file_text": False, "history_message":raw_input.get("history_message", ""), "extracted_text": "", "image_description": "", "asr_text": "", "file_text": raw_input.get("file_text", ""), "combined_query": "", "task_type": "", "task_classification_result": None, "workflow_result": None, "final_response": "", "error_message": "", "route_flag": initial_route_flag, "route_params": {} } main_result = await main_agent.ainvoke(initial_state) route_flag_value = (main_result.get("route_flag")or main_result.get("task_type")or "") route_params = main_result.get("route_params") or {"combined_query": main_result.get("combined_query", "") or main_result.get("extracted_text", ""),"extracted_text": main_result.get("extracted_text", "")} if "file_text" in raw_input: route_params["file_text"] = raw_input["file_text"] if "history_message" not in route_params and "history_message" in raw_input: route_params["history_message"] = filter_image_urls(raw_input["history_message"]) print("历史记录",route_params["history_message"]) print("用户问题", route_params["combined_query"]) print("ttttttt", route_params["extracted_text"]) sub_result = await execute_workflow_chain( route_flag=route_flag_value, route_params=route_params, chat_id=chat_id, message_id=message_id, checkpointer=checkpointer ) processed_content = "" if sub_result.get("response"): processed_content = sub_result.get("response", "") else: logger.warning("子agent没有返回response") processed_content = "" final_result = { "content": processed_content, "actions": sub_result.get("actions", []), "result_tag": sub_result.get("result_tag", ""), "suggestedReplies": sub_result.get("suggestedReplies", []) } except Exception as e: execution_error = e finally: execution_done.set() await event_queue.put({"type": "execution_complete"}) agent_task = asyncio.create_task(run_agent()) execution_complete_received = False rag_result = [] graph_result = [] while True: try: event = await asyncio.wait_for(event_queue.get(), timeout=0.1) if event.get("type") == "execution_complete": execution_complete_received = True continue event_type = event.get("type") title = event.get("title", "") details = event.get("details", "") result = event.get("result") if event_type == "function_execution": logger.info(f"function_execution event - title: {title}, has_result: {result is not None}, result_type: {type(result)}") if title == "知识库搜索工具" and result and isinstance(result, dict): print("===============================", result) rag_result_raw = result.get("sourceCitation", {}) rag_result = {} for doc_name, chunks in rag_result_raw.items(): if isinstance(chunks, list): rag_result[doc_name] = [ {k: v for k, v in item.items() if k != "text"} if isinstance(item, dict) else item for item in chunks ] else: rag_result[doc_name] = chunks elif title == "图谱检索工具" and result and isinstance(result, dict): print("===============================", result) graph_result = result.get("xxxx.graph", []) else: yield format_stream_event("function_end", { "title": title, "details": details }, chat_id=chat_id, message_id=message_id) elif event_type == "function_error": yield format_stream_event("function_error", { "title": title, "error": details, "details": details }, chat_id=chat_id, message_id=message_id) elif event_type == "stream_content": content = event.get("content", "") yield format_stream_event("stream_content", { "content": content }, chat_id=chat_id, message_id=message_id) except asyncio.TimeoutError: if execution_complete_received: break if execution_done.is_set() and event_queue.empty(): break continue await agent_task if execution_error: yield format_stream_event("error", { "error": str(execution_error) }, chat_id=chat_id, message_id=message_id) else: title = await extract_conversation_title(final_result.get("content", "") if final_result else "") yield format_stream_event("execution_complete", { "final_result": final_result }, chat_id=chat_id, message_id=message_id) yield format_stream_event("execution_complete", { "rag_result": rag_result, "graph_result": graph_result, "title": title }, chat_id=chat_id, message_id=message_id) except Exception as e: yield format_stream_event("error", { "error": str(e) }, chat_id=chat_id, message_id=message_id) finally: unregister_event_callback(stream_handler) clear_current_stream_handler() # =============== API 接口 =============== @app.post("/api/agent/stream_process") async def stream_process_request(request: AgentRequest): """ 流式处理 Agent 请求，实时输出节点执行信息 不保存历史，调用端会自己保存历史对话信息 每次请求时创建新的主Agent实例 """ print(f"🔍 [调试] 接收到的原始请求: {request.model_dump_json(indent=2)}") print(f"[DEBUG] 收到请求 chat_id={request.chat_id!r}, message_id={request.message_id!r}, route_flag={request.route_flag!r}, query={str(request.query)[:100]}") # 1. 构建当前请求的输入数据 raw_input = {} if request.query: if isinstance(request.query, str): raw_input["query"] = request.query elif isinstance(request.query, list): for item in request.query: if item.get("type") == "text": raw_input["query"] = item.get("text") print(f"当前查询: {raw_input['query']}") if request.image: raw_input["image"] = request.image if request.audio: # audio 字段支持 audio 或 asr，统一使用 audio 键 raw_input["audio"] = request.audio # [新增] 提取 file_text if request.file_text: raw_input["file_text"] = request.file_text[:5000] # 历史消息预处理：解析、移除当前用户消息、序列化 if request.history_message: raw_input["history_message"] = preprocess_from_request(request.history_message, exclude_last=True) # 2. 验证至少有一个输入 if not raw_input: raise HTTPException( status_code=400, detail="至少需要提供一个输入：query、image 或 audio" ) # 3. 获取路由标志（如果提供，将跳过任务分类步骤） # 始终传递 route_flag，默认值为空字符串 route_flag = request.route_flag or "" print("打印agent对应的route_flag：") print(route_flag) rag_prompt = process_rag_prompt(request.rag_prompt) if request.rag_prompt else "" chat_id = request.chat_id or "" message_id = request.message_id or "" # 记录智能体使用情况 if route_flag and chat_id: from tools.agent_usage_statistics import record_agent_usage asyncio.create_task(record_agent_usage( chat_id=chat_id, route_flag=route_flag, message_id=message_id )) logger.info(f"记录智能体使用: chat_id={chat_id}, route_flag={route_flag}") # 4. 设置当前请求的用户配置和知识库参数（线程安全，支持多用户并发） # 如果请求中提供了这些参数，则使用请求中的值；否则使用默认值 set_request_user_config( x_user_id=request.x_user_id, x_user_name=request.x_user_name, x_role=request.x_role, text_kb_id=request.text_kb_id, text_file_id=request.text_file_id ) #5. 执行主Agent async def event_generator(): async for event in stream_main_agent_execution( raw_input=raw_input, route_flag=route_flag, rag_prompt=rag_prompt, chat_id=chat_id, message_id=message_id ): yield event return StreamingResponse(event_generator(), media_type="text/event-stream") @app.get("/api/download/{file_name}") async def download_file(file_name: str): """ 文件下载接口 Args: file_name: 文件名 Returns: 文件下载响应 """ pdf_path = os.path.join(PDF_DIR, file_name) word_path = os.path.join(WORD_DIR, file_name) file_path = pdf_path if os.path.exists(pdf_path) else word_path print(f"下载文件：{file_path}") # 检查文件是否存在 ext = os.path.splitext(file_name)[1].lower() media_type = { ".pdf": "application/pdf", ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ".doc": "application/msword", }.get(ext, "application/octet-stream") # 返回文件 return FileResponse( path=file_path, filename=file_name, media_type=media_type ) @app.get("/api/health") async def health_check(): """健康检查接口""" return { "status": "healthy", "message": "服务正常运行，每次请求时创建新的Agent实例" } @app.post("/api/v1/feedback/classify") async def feedback_classify(request: FeedbackClassifyRequest): """ 反馈内容分类接口 功能： 1. 从反馈文本中提取设备名称 2. 根据设备名称反推系统名称 3. 对反馈内容进行分类 Args: request: 包含 feedback_text 字段的请求体 Returns: { "success": true, "system": "系统名称", "category": "分类结果" } 分类类别： - 信息过时：反映文档或手册中的信息已经过时，不再适用 - 步骤错误：反映操作步骤或流程存在错误 - 步骤缺失：反映缺少必要的操作步骤或流程说明 - 备件信息有误：反映备件型号、规格、数量等信息有误 - 安全提示不足：反映缺少必要的安全警示或注意事项 - 图示不清：反映图片、示意图不清晰或难以理解 - 其他：不属于以上类别 """ from tools.fault_statistics import analyze_feedback_and_classify result = await analyze_feedback_and_classify(request.feedback_text) if not result.get("success", False): raise HTTPException( status_code=500, detail=result.get("error", "分类失败") ) return result class PdfContentItem(BaseModel): page_idx: int bbox: Optional[str] = None text: Optional[str] = None text_level: Optional[int] = None img_path: Optional[str] = None table_body: Optional[str] = None class RequestWrapper(BaseModel): pdf_contents: List[PdfContentItem] @app.post("/split_content_list") async def split_content_list(request_data: RequestWrapper): pdf_contents = request_data.pdf_contents processed_list = [] for item in pdf_contents: # 将 Pydantic 模型转换为字典，方便修改 item_dict = item.model_dump() # --- 核心逻辑：判断并添加 type 字段 --- item_type = None # 1. 如果 text 非空 -> type: "text" if item.text: item_type = "text" # 2. 如果 table_body 非空 -> type: "table" # 注意：如果 text 和 table_body 都有，根据代码顺序，table 会覆盖 text。 # 如果你的业务逻辑是互斥的或需要优先级，请调整此处顺序。 elif item.table_body: item_type = "table" # 3. 如果 img_path 非空 且 table_body 为空 -> type: "image" # 注意：上面用了 elif 判断 table_body，所以到这里 table_body 肯定为空， # 但为了逻辑清晰，显式写出条件。 elif item.img_path and not item.table_body: item_type = "image" # 如果匹配到了类型，写入字典 if item_type: item_dict["type"] = item_type processed_list.append(item_dict) slices = get_chunk_bbox(processed_list) slices_check = chunk_check(slices, 8000) return {"code": 200, "message": "ok", "data": {"slices": slices_check}} @app.post("/split_result") async def split_result( file: UploadFile = File(...), image_prefix: str = Form("/api/v1/knowledge/files/images/"), ): if not file.filename: raise HTTPException(status_code=400, detail="File name is missing") chunk_size = 1024 chunk_overlap = 100 filename = file.filename suffix = PathLib(filename).suffix.lower() # 获取文件后缀 temp_input_path: Optional[PathLib] = None converted_pdf_path: Optional[PathLib] = None logger.info("调用知识切分接口") try: # ? 创建临时文件时指定后缀名 with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp_in: content = await file.read() logger.info(f"文件大小:{len(content)}") logger.info(f"前10字节:{content[:10]}") # if not content.startwith(b"%PDF"): # logger.error("不是合法PDF") # raise HTTPException(status_code=400,detail="Uploaded file is not a valid PDF") tmp_in.write(content) temp_input_path = PathLib(tmp_in.name) # 2. 根据文件类型处理 if suffix == ".pdf": logger.info("调用pdf切分方式") result_data = await process_pdf_file(temp_input_path, image_prefix,filename) if result_data is None: raise HTTPException(status_code=502, detail="PDF processing failed") elif suffix == ".docx": logger.info("调用docx切分方式") logger.info(f"转换 DOCX -> PDF: {filename}") # 执行转换 converter.convert(str(temp_input_path), output_dir=str(DATA_DIR)) # ? 使用临时文件的基础名查找 PDF temp_base_name = temp_input_path.stem # 如 tmp2bb0l4cn converted_pdf_path = PathLib(DATA_DIR) / f"{temp_base_name}.pdf" if not converted_pdf_path.exists(): raise HTTPException(status_code=500, detail=f"PDF 转换失败: {converted_pdf_path}") result_data = await process_pdf_file(converted_pdf_path, image_prefix) if result_data is None: raise HTTPException(status_code=502, detail="DOCX to PDF processing failed") elif suffix in (".txt", ".md"): logger.info("调用txt切分方式") # 直接读取文本内容 async with aiofiles.open(temp_input_path, "r", encoding="utf-8", errors="ignore") as f: text_content = await f.read() contents = split_text_preserve_sentences(text_content) # 构造统一结构（无 images） slices = [] for ins in contents: slices.append({"content": ins, "positions": []}) result_data = {"slices": slices, "images": {}} elif suffix in [".xlsx"]: # converter.convert(str(temp_input_path), output_dir=str(DATA_DIR)) # temp_base_name = temp_input_path.stem # 如 tmp2bb0l4cn # converted_pdf_path = PathLib(DATA_DIR) / f"{temp_base_name}.pdf" # if not converted_pdf_path.exists(): # raise HTTPException(status_code=500, detail=f"PDF 转换失败: {converted_pdf_path}") logger.info("调用xlsx切分方式") result_data = await process_other_file(temp_input_path, image_prefix,filename) if result_data is None: raise HTTPException(status_code=502, detail="DOCX parse failed") elif suffix in [".ppt", ".pptx"]: logger.info("调用 PPT/PPTX 切分方式") # 1. 判断是否为旧版 .ppt 格式，如果是，则先进行转换 if suffix == ".ppt": logger.info(f"检测到旧版 PPT 文件，正在转换为 PPTX: {temp_input_path}") try: # 定义 LibreOffice 转换命令 command = [ 'libreoffice', '--headless', # 无头模式，不弹出图形界面 '--convert-to', 'pptx', '--outdir', os.path.dirname(temp_input_path), temp_input_path ] # 2. 使用 run_in_executor 将阻塞的同步转换操作放入线程池，避免阻塞异步事件循环 loop = asyncio.get_event_loop() result = await loop.run_in_executor( None, lambda: subprocess.run(command, capture_output=True, text=True) ) # 3. 检查转换是否成功 if result.returncode != 0: logger.error(f"PPT 转换失败: {result.stderr}") raise HTTPException(status_code=500, detail="PPT to PPTX conversion failed") # 4. 转换成功后，更新输入路径为新生成的 .pptx 文件路径 temp_input_path = os.path.splitext(temp_input_path)[0] + ".pptx" logger.info(f"PPT 转换成功，新文件路径: {temp_input_path}") except Exception as e: logger.exception(f"PPT 转换过程发生异常: {e}") raise HTTPException(status_code=500, detail=f"PPT conversion error: {str(e)}") # 5. 无论是原本就是 .pptx，还是刚刚转换完成的 .pptx，都统一走后续处理逻辑 result_data = await process_other_file(temp_input_path, image_prefix, filename) if result_data is None: raise HTTPException(status_code=502, detail="PPTX parse failed") else: raise HTTPException( status_code=400, detail="Unsupported file type. Only PDF, DOCX, TXT, and MD are supported." ) return JSONResponse({"code": 200, "message": "ok", "data": result_data}) except requests.RequestException as e: logger.error(f"Split service request failed: {e}") raise HTTPException(status_code=502, detail=f"Split service error: {str(e)}") except Exception as e: logger.error(f"Internal error: {e}") raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}") finally: # 清理临时文件 for path in [temp_input_path, converted_pdf_path]: if path and path.exists(): try: path.unlink() except Exception as e: logger.warning(f"Failed to delete temp file {path}: {e}") @app.get("/") async def root(): """根路径""" return { "message": "Agent API 服务", "docs": "/docs" } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=9090)