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") @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 StreamingResponse( stream_llm_content(prompt), media_type="text/plain; charset=utf-8", ) 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 def length_convert(length: Optional[str]) -> str: if length == "短": return "1000字" if length == "中": return "1000-2000字" if length == "长": return "2000-5000字" return length or "" def build_write_prompt(request: WriteRequest) -> str: if request.prompt and request.prompt.strip(): return request.prompt.strip() template = request.template or "" 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不能为空") return prompt_template.format( role=request.role or "", title=title, length=length_convert(request.length), requirement=request.requirement or "", references=request.references or "", outline=request.outline or "", ) async def stream_write_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("/Write") async def write(request: WriteRequest): prompt = build_write_prompt(request) return StreamingResponse( stream_write_content(prompt), media_type="text/plain; charset=utf-8", ) 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_convert(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 StreamingResponse( stream_outline_content(prompt), media_type="text/plain; charset=utf-8", ) 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): conten1 = """ 人工智能正在深刻改变现,代教育的方式。通过智能教学系统【】AI可以根据学生的学习习惯和掌握情况，提供个性化的学习内容和节奏学习内容和节奏，提高学习效率。同时，AI还能辅助教师进行作业批改、学习分析，甚至参与课堂互动，减轻教师的重复工作压力。 虚拟助教、智能语音识别、自动文达系统等技术，也让远程教育更加高效和便捷。更加高效和便捷。随着AI技术的发展，教育不再局限于固定的教材和课堂，而是变得更加开放、灵活和智能化------让每个雪生都能获得更适合自己的学习体验。人工智能不仅是工具，更是推动教育公平和质量提升的重要力量。 """ reasoning1 = """ "深刻改变现,代教育中"多了","，修改为“深刻改变现代教育中”；“智能教学系统【】”中存在错误符号“【】”，修改为“智能教学系统”；“提供个性化的学习内容和节奏学习内容和节奏”存在重复内容，修改为"提供个性化的学习内容和节奏"；"自动文达系统"存在错误词语"文达"，修改为”自动问答系统“；”而是变得更加开放、灵活和智能化------“ 中存在错误符号“------”，修改为“而是变得更加开放、灵活和智能化，”；“每个雪生”中存在错别字“雪生”，修改为“每个学生”；检查完上述错误之后，正确的文本内容应该是：人工智能正在深刻改变现代教育的方式。通过智能教学系统，AI可以根据学生的学习习惯和掌握情况，提供个性化的学习内容和节奏，提高学习效率。同时，AI还能辅助教师进行作业批改、学习分析，甚至参与课堂互动，减轻教师的重复工作压力。虚拟助教、智能语音识别、自动问答系统等技术，也让远程教育更加高效和便捷。 随着AI技术的发展，教育不再局限于固定的教材和课堂，而是变得更加开放、灵活和智能化，让每个学生都能获得更适合自己的学习体验。人工智能不仅是工具，更是推动教育公平和质量提升的重要力量。 """ ques = f""" 文本内容为{conten1},对给出的文本内容进行逻辑校对、基础校对和合规性检查，逻辑校对检测文本逻辑连贯性，基础校对检查文本中错别字、错误标点符号和重复内容。 校对文本之后，如果你发现没有任何错误，不需要返回其他内容，只需要返回： ###\n### 发现错误时返回： ###\n错误：错别字、标点不规范\n原文：星光店电；万里乌云\n建议：星光点点，万里无云 \n错误：错别词语、标点不规范\n原文：请注丰收与团圆。\n建议：庆祝丰收与团圆。 \n错误：标点不规范\n原文：你好；明天\n建议：你好，明天 \n错误：重复内容\n原文：我听说你心情很不错不错。你心情很不错不错。\n建议：我听说你心情很不错。### """ conten2 = """ 关于近期工作安排的通知 各位同事， 京公司研九决定，现将近期工作安排通知通知如下： 一、项目推进方面。各部门需加快项目进度，确保按时交付。如有问题，及时与相关领导沟通解决。项目负责人要切实履行职责，加强团队协作，提高工作效率。 二、培训学习方面：公司将组织系列培训课程，提升员工专业技能。具体时间、地点另行通知。员工需按时参加，不得无辜缺席。自即日起，严禁在工作时间从事与工作无关的活动。 二、纪律要求。近期发现部分员工上班时间做与工作无关的事情，如炒股、聊天等，严重影响工作效率和公司形象。自即日起，严禁在工作时间从事与工作无关的活动。违反规定者，将按照公司规章制度严肃处理。 特此通知： [公司名称] [日期] """ reasoning2 = """ "京公司研九决定"中存在错别字和词语“京”、“研九”，修改为“经公司研究决定”；“工作安排通知通知”存在重复内容，修改为“工作安排通知”；“一、项目推进方面。”存在错误标点符号，参与后续内容中“二、培训学习方面：”，因此“一、项目推进方面”之后应当使用“：”，修改为“一、项目推进方面：”；“无辜缺席。”中存在错别字“辜”，修改为“无故缺席”； “二、纪律要求。”存在错别字、标点不规范，纪律要求应该是第三点，应该使用”：“，修改为“三、纪律要求：”；”特此通知：“中存在标点不规范，应该修改为”特此通知。“检查完上述错误之后，正确的文本内容应该是：关于近期工作安排的通知 各位同事， 经公司研究决定，现将近期工作安排通知如下： 一、项目推进方面：各部门需加快项目进度，确保按时交付。如有问题，及时与相关领导沟通解决。项目负责人要切实履行职责，加强团队协作，提高工作效率。 二、培训学习方面：公司将组织系列培训课程，提升员工专业技能。具体时间、地点另行通知。员工需按时参加，不得无故缺席。自即日起，严禁在工作时间从事与工作无关的活动。 三、纪律要求：近期发现部分员工上班时间做与工作无关的事情，如炒股、聊天等，严重影响工作效率和公司形象。自即日起，严禁在工作时间从事与工作无关的活动。违反规定者，将按照公司规章制度严肃处理。 特此通知。 [公司名称] [日期] """ ques2 = f""" 文本内容为{conten2},对给出的文本内容进行逻辑校对、基础校对和合规性检查，逻辑校对检测文本逻辑连贯性，基础校对检查文本中错别字、错误标点符号和重复内容。 校对文本之后，如果你发现没有任何错误，不需要返回其他内容，只需要返回： ###\n### 发现错误时返回： ###\n错误：错别字、标点不规范\n原文：星光店电；万里乌云\n建议：星光点点，万里无云 \n错误：错别词语、标点不规范\n原文：请注丰收与团圆。\n建议：庆祝丰收与团圆。 \n错误：标点不规范\n原文：你好；明天\n建议：你好，明天 \n错误：重复内容\n原文：我听说你心情很不错不错。你心情很不错不错。\n建议：我听说你心情很不错。### """ conten3 = """ 《水声目标识别》课程教学计划 一、课程名称 水声目标识别，该课程式用于海洋科学、水声工程、电子信息等专业本科生asfqfqfqwfwqf、研究生，以及军事院校学员、相关领域科研人员和企业技术人员技术人员。 三、计划学时 2026学年，总学时：48学时（理论授课32学时，实作授课16学时） 三、授课时间 每周4学时，共12周完成 四、教学目标 1. 知识目标 使学生掌握水声目标识别的几本原理】方法和技术，包括声纳信号的特性、目标特征提取、分类与识别算法等。 了解水声目标识别在军事（如潜艇作战、反潜作战等）和民用（如海洋资源勘探、水下考古等）领域的应用。 熟悉水声目标识别系统的基本组成和工作原理， 2. 能力目标 培养学生运用所学知识对水声信号进行分析处理、特征提取和目标识别的能力。 提高学生运用专业软件（如MATLAB等）进行水声水声信号处理和目标识别仿真的能力。 增强学生解决实际水声目标识别问题的实践能力和创新思维。 3. 素质目标 培养学生严谨的科学态度和扎实的专业素养，使其剧备良好的团队协作精神和沟通能力。 激发学生对水声技术领域的学习兴趣和探索精神，培养其自主学习和终身学习的意识。 """ ques3 = f""" 文本内容为{conten3},对给出的文本内容进行逻辑校对、基础校对和合规性检查，逻辑校对检测文本逻辑连贯性，基础校对检查文本中错别字、错误标点符号和重复内容。 校对文本之后，如果你发现没有任何错误，不需要返回其他内容，只需要返回： ###\n### 发现错误时返回： ###\n错误：错别字、标点不规范\n原文：星光店电；万里乌云\n建议：星光点点，万里无云 \n错误：错别词语、标点不规范\n原文：请注丰收与团圆。\n建议：庆祝丰收与团圆。 \n错误：标点不规范\n原文：你好；明天\n建议：你好，明天 \n错误：重复内容\n原文：我听说你心情很不错不错。你心情很不错不错。\n建议：我听说你心情很不错。### """ reasoning3 = """ "式用"中存在错别字“式”，修改为：“适用”；“本科生asfqfqfqwfwqf、”中存在错别字和符号“asfqfqfqwfwqf”，修改为：“本科生、”；“三、计划学时”存在错别字使用不当，修改为：“二、计划学时”； “几本原理】”存在错别字、标点不规范，应该修改为：“基本原理”；“工作原理，”存在标点不规范，应该修改为：“工作原理。”；“水声水声信号”中存在重复内容“水声”，应该修改为“水声信号”； “剧备”存在错别字“剧”，应该修改为：”具备“。检查完上述错误之后，正确的文本内容应该是： 《水声目标识别》课程教学计划 一、课程名称 水声目标识别，该课程式用于海洋科学、水声工程、电子信息等专业本科生、研究生，以及军事院校学员、相关领域科研人员和企业技术人员。 三、计划学时 2026学年，总学时：48学时（理论授课32学时，实作授课16学时） 三、授课时间 每周4学时，共12周完成 四、教学目标 1. 知识目标 使学生掌握水声目标识别的基本原理、方法和技术，包括声纳信号的特性、目标特征提取、分类与识别算法等。 了解水声目标识别在军事（如潜艇作战、反潜作战等）和民用（如海洋资源勘探、水下考古等）领域的应用。 熟悉水声目标识别系统的基本组成和工作原理。 2. 能力目标 培养学生运用所学知识对水声信号进行分析处理、特征提取和目标识别的能力。 提高学生运用专业软件（如MATLAB等）进行水声信号处理和目标识别仿真的能力。 增强学生解决实际水声目标识别问题的实践能力和创新思维。 3. 素质目标 培养学生严谨的科学态度和扎实的专业素养，使其具备良好的团队协作精神和沟通能力。 激发学生对水声技术领域的学习兴趣和探索精神，培养其自主学习和终身学习的意识。 """ # 定义 Few-Shot CoT 示例 examples = [ { "question": ques, "reasoning": reasoning1, "answer": """ \n错误：标点不规范\n原文：现,代教育\n建议：现代教育 \n错误：标点不规范\n原文：智能教学系统【】\n建议：智能教学系统 \n错误：重复内容\n原文：提供个性化的学习内容和节奏学习内容和节奏\n建议：提供个性化的学习内容和节奏 \n错误：错别字、错别词语\n原文：自动文达系统\n建议：自动问答系统 \n错误：标点不规范\n原文：而是变得更加开放、灵活和智能化------\n建议：而是变得更加开放、灵活和智能化， \n错误：错别字、错别词语\n原文：每个雪生\n建议：每个学生 """ }, { "question": ques2, "reasoning": reasoning2, "answer": """ \n错误：错别字、错别词语\n原文：京公司研九决定\n建议：经公司研究决定 \n错误：重复内容\n原文：工作安排通知通知\n建议：工作安排通知 \n错误：错别字、错别词语\n原文：无辜缺席。\n建议：无故缺席 \n错误：错别字、标点不规范\n原文：二、纪律要求。\n建议：三、纪律要求： \n错误：标点不规范\n原文：特此通知：\n建议：特此通知。 """ }, { "question": ques3, "reasoning": reasoning3, "answer": """ \n错误：错别字、错别词语\n原文：式用\n建议：适用 \n错误：错别字、错别字符\n原文：本科生asfqfqfqwfwqf、\n建议：本科生、 \n错误：错别字、逻辑不顺\n原文：三、计划学时\n建议：二、计划学时 \n错误：错别字、标点不规范\n原文：几本原理】\n建议：基本原理 \n错误：标点不规范\n原文：工作原理，\n建议：工作原理。 \n错误：重复内容\n原文：水声水声信号\n建议：水声信号 \n错误：错别字、错别词语\n原文：剧备\n建议：具备 """ } ] # 定义单个示例的模板 example_template = """ 问题: {question} 推理过程: {reasoning} 答案: {answer} """ # 创建 PromptTemplate 对象 example_prompt = PromptTemplate( input_variables=["question", "reasoning", "answer"], template=example_template ) # 创建 FewShotPromptTemplate 对象 few_shot_prompt = FewShotPromptTemplate( examples=examples, example_prompt=example_prompt, prefix="你是一个文本内容校对高手。能够根据要要求进行错误检查。请根据以下示例，解决用户提出的问题。", suffix="问题: {question}\n推理过程:", input_variables=["question"], example_separator="\n\n" ) prompt = few_shot_prompt.format(question=query) return prompt def build_review_prompt(types: List[str], content: str, require: Optional[str]) -> Iterable[ChatCompletionMessageParam]: if require is None: require = "" 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" query = f""" 我会给你对应的文本，进行相应的检查，{review_t} """ query1 = cotprompt(content) return [ ChatCompletionUserMessageParam(role="user", content="请不要进行思考，直接输出内容"), 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 async def call_review_model(prompt: Iterable[ChatCompletionMessageParam]) -> str: system_prompt = ( "你是一个专业的文本校对专家，能够根据要求进行内容的纠正和改错" "请严格按照用户要求的格式输出。" "不要输出推理过程、解释性文字、前缀或总结。" ) full_text = "" async for chunk in model_api.OpenaiAPI.open_api_chat_stream( model=None, json_output=False, system_prompt=system_prompt, messages=prompt, enable_thinking=False, temperature=0.1, ): if chunk: full_text += chunk return full_text 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 = 1000 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", response_model=ReviewResponse) async def review(request: ReviewRequest): result = await review_text( content=request.content, types=request.types, require=request.require, sensitive_words=request.sensitive_words, negative_words=request.negative_words, positive_words=request.positive_words, ) return {"data": result} # =============== 请求模型 =============== 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)