2175 lines
78 KiB
Python
2175 lines
78 KiB
Python
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())
|
||
|
||
try:
|
||
import ahocorasick
|
||
except ImportError:
|
||
ahocorasick = None
|
||
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:
|
||
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
|
||
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,
|
||
safe_original_filename,
|
||
image_base64_to_data_url,
|
||
extract_image_text,
|
||
prepare_split_image,
|
||
)
|
||
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
|
||
import unicodedata
|
||
from difflib import SequenceMatcher
|
||
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
|
||
)
|
||
from gw_write import (
|
||
REVIEW_TYPE_ERROR_MAP,
|
||
cotprompt,
|
||
build_review_prompt,
|
||
length_convert,
|
||
length_display,
|
||
build_length_instruction,
|
||
dedupe_and_filter,
|
||
filter_positive_word_false_positives,
|
||
append_dictionary_results,
|
||
find_keywords,
|
||
normalize_words,
|
||
process_rag_prompt,
|
||
get_allowed_review_error_types,
|
||
normalize_review_types,
|
||
)
|
||
app = FastAPI(max_request_size=1024 * 1024 * 10)
|
||
|
||
_doc2pdf_converter: Optional[Doc2PDF] = None
|
||
DATA_DIR = "/app/files"
|
||
load_dotenv()
|
||
VLM_SEMAPHORE = asyncio.Semaphore(1)
|
||
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger(__name__)
|
||
REWRITE_TYPE_MAP = {
|
||
"quotePolish": "金句润色",
|
||
"formalStyle": "书面化",
|
||
"speechStyle": "讲稿化",
|
||
"formatting": "规整",
|
||
"expand": "扩写",
|
||
"continueWriting": "续写",
|
||
"report": "汇报",
|
||
"quote": "金句",
|
||
"summarize": "精简",
|
||
"condense": "总结",
|
||
}
|
||
# /Review 流式并发配置：块大小 5000，最多 5 块并行
|
||
REVIEW_CHUNK_SIZE = 1000
|
||
REVIEW_MAX_CONCURRENCY = 5
|
||
REVIEW_SYSTEM_PROMPT = (
|
||
"你是一个严谨的专业文本校对专家，必须完整检查用户明确选中的校对范围，并返回该范围内的真实问题。"
|
||
"禁止检查或输出用户未选中的校对类别，禁止重复输出相同问题，原文与建议相同时不得输出。"
|
||
"校对结果会由程序直接替换进正式公文，因此建议字段只能包含最终正文，任何解释、括号备注或操作说明都会污染公文，绝对禁止输出。"
|
||
"每条原文字段必须是输入正文中真实存在的连续原样子串，不能包含任何提前修改后的文字，否则该结果会被系统丢弃。"
|
||
"请严格按照用户要求的格式输出。"
|
||
"不要输出推理过程、解释性文字、前缀或总结。"
|
||
)
|
||
def get_doc2pdf_converter() -> Doc2PDF:
|
||
global _doc2pdf_converter
|
||
if _doc2pdf_converter is None:
|
||
try:
|
||
_doc2pdf_converter = Doc2PDF()
|
||
except RuntimeError as exc:
|
||
logger.error(f"LibreOffice 初始化失败: {exc}")
|
||
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
||
return _doc2pdf_converter
|
||
|
||
|
||
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
|
||
|
||
|
||
|
||
|
||
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}")
|
||
|
||
target_text = (request.sentence or request.content or "").strip()
|
||
if not target_text:
|
||
raise HTTPException(status_code=400, detail="缺少需要改写的内容")
|
||
|
||
# 润色区的四种能力都是编辑器选区级操作。这里不传入全文上下文，避免模型
|
||
# 把 content 误判为改写对象，从而在只选择一句话时仍输出整篇公文。
|
||
selection_style_instructions = {
|
||
"金句润色": "在保持原意的基础上提炼语言，使表达精练、有力、有文采",
|
||
"书面化": "改写为严谨、规范、庄重的公文书面表达",
|
||
"讲稿化": "改写为适合正式场合表达、自然连贯的讲稿语言",
|
||
"规整": "改写为格式规范、内容严谨、语言得体的公文表达",
|
||
}
|
||
if beautify_type in selection_style_instructions:
|
||
style_instruction = selection_style_instructions[beautify_type]
|
||
return f"""你是专业的公文局部改写助手。
|
||
下面“待改写内容”是本次唯一允许处理和输出的内容。
|
||
请{style_instruction}，保持原意和事实信息，不增加新的时间、地点、人物、数据或事项。
|
||
不得扩写为完整公文、完整讲稿或完整段落，不得补充标题、称谓、开场白、结尾或上下文。
|
||
输出长度应与待改写内容大致相当。
|
||
禁止使用Markdown，禁止解释修改过程，禁止复述未被选中的任何内容。
|
||
只输出改写结果。
|
||
|
||
其他要求：{request.require or "无"}
|
||
待改写内容：
|
||
<selection>
|
||
{target_text}
|
||
</selection>"""
|
||
|
||
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=target_text,
|
||
)
|
||
|
||
|
||
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
|
||
|
||
|
||
WRITE_FORMULA_INSTRUCTION = r"""
|
||
|
||
【公式输出规范（必须遵守）】
|
||
1. 仅在正文确实需要数学、统计或技术公式时使用 LaTeX；普通数字、百分比、日期、编号和金额均使用普通文本，不要为了排版而生成公式。
|
||
2. 行内公式必须且只能写成 `$公式内容$`。开始和结束的 `$` 之间不得换行。
|
||
3. 独占一行的公式必须使用成对的 `$$`，公式内容放在两者之间。
|
||
4. 只输出 KaTeX 支持的标准 LaTeX。禁止使用 `\(...\)`、`\[...\]`、LaTeX 文档环境、代码块、HTML、MathML 或 JSON 包裹公式。
|
||
5. 每个公式的定界符必须成对闭合。公式中的说明文字使用 `\text{...}`；百分号写作 `\%`，乘号优先写作 `\times`。
|
||
6. 金额直接写成“100元”“20万元”等普通文本，不得使用 `$` 作为货币符号。
|
||
7. 若正文不需要公式，不要输出任何 `$` 或 `$$`。
|
||
"""
|
||
|
||
WRITE_HEADING_LAYOUT_INSTRUCTION = """
|
||
|
||
【标题与正文换行规范（必须遵守）】
|
||
1. “一、……”形式的一级标题和“（一）……”形式的二级标题必须分别独占一行，严禁把一级标题、二级标题或正文连续写在同一行。
|
||
2. 每个二级标题后必须换行，正文从下一段开始；正文结束后再另起一行输出下一个二级标题。
|
||
3. 标题与正文之间、相邻层级标题之间使用一个空行分隔，以便前端 Markdown 正确显示。
|
||
4. 必须按以下结构输出，不得写成“（一）标题 正文”：
|
||
|
||
一、一级标题
|
||
|
||
（一）二级标题
|
||
|
||
正文内容。
|
||
|
||
（二）二级标题
|
||
|
||
正文内容。
|
||
"""
|
||
|
||
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 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_HEADING_LAYOUT_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不能为空")
|
||
normalized_types = normalize_review_types(self.types)
|
||
unknown_types = [
|
||
review_type for review_type in self.types
|
||
if review_type not in REVIEW_TYPE_ERROR_MAP
|
||
]
|
||
if unknown_types:
|
||
raise ValueError(f"不支持的检查类型：{'、'.join(dict.fromkeys(unknown_types))}")
|
||
if not normalized_types:
|
||
raise ValueError("types不能为空")
|
||
self.types = normalized_types
|
||
return self
|
||
|
||
|
||
class ReviewItem(BaseModel):
|
||
original: str
|
||
error: str
|
||
suggestion: str
|
||
|
||
|
||
class ReviewResponse(BaseModel):
|
||
data: List[ReviewItem]
|
||
|
||
|
||
|
||
|
||
|
||
_REVIEW_BLOCK_RE = re.compile(
|
||
r"错误[:：]\s*(?P<error>.*?)\s*[\r\n]+"
|
||
r"原文[:：]\s*(?P<original>.*?)\s*[\r\n]+"
|
||
r"建议[:：]\s*(?P<suggestion>.*?)(?=\n\s*\n|$)",
|
||
re.S,
|
||
)
|
||
_COMPLETED_REVIEW_BLOCK_RE = re.compile(
|
||
r"错误[:：]\s*(?P<error>.*?)\s*[\r\n]+"
|
||
r"原文[:：]\s*(?P<original>.*?)\s*[\r\n]+"
|
||
r"建议[:：]\s*(?P<suggestion>.*?)(?=\r?\n\s*\r?\n|\r?\n?###)",
|
||
re.S,
|
||
)
|
||
|
||
def parse_review_output(text: str) -> List[dict]:
|
||
text = text.strip()
|
||
|
||
if text.startswith("###"):
|
||
text = text[3:].strip()
|
||
|
||
if text.endswith("###"):
|
||
text = text[:-3].strip()
|
||
|
||
result = []
|
||
|
||
for match in _REVIEW_BLOCK_RE.finditer(text):
|
||
item = {
|
||
"error": match.group("error").strip(),
|
||
"original": match.group("original").strip(),
|
||
"suggestion": match.group("suggestion").strip(),
|
||
}
|
||
# 原文字段为空或仅包含空白时，不作为有效校对记录。
|
||
if item["original"]:
|
||
result.append(item)
|
||
|
||
return result
|
||
|
||
|
||
def _has_repeating_review_tail(text: str) -> bool:
|
||
"""检测模型是否开始连续重复同一条完整记录。"""
|
||
matches = list(_COMPLETED_REVIEW_BLOCK_RE.finditer(text))
|
||
if len(matches) < 2:
|
||
return False
|
||
|
||
def record_key(match) -> tuple:
|
||
return (
|
||
match.group("error").strip(),
|
||
match.group("original").strip(),
|
||
match.group("suggestion").strip(),
|
||
)
|
||
|
||
return record_key(matches[-1]) == record_key(matches[-2])
|
||
|
||
|
||
def _split_review_error_types(error: str) -> List[str]:
|
||
"""按原顺序拆分并去重校对错误类型。"""
|
||
aliases = {
|
||
"逻辑错误": "逻辑不通",
|
||
"逻辑不顺": "逻辑不通",
|
||
"错别字符": "错别字",
|
||
}
|
||
result = []
|
||
for error_type in re.split(r"[、,，]", error or ""):
|
||
error_type = aliases.get(error_type.strip(), error_type.strip())
|
||
if error_type and error_type not in result:
|
||
result.append(error_type)
|
||
return result
|
||
|
||
|
||
def _is_whitespace_only_change(original: str, suggestion: str) -> bool:
|
||
"""判断原文与建议是否相同，或仅存在空白差异。"""
|
||
if not original or not suggestion:
|
||
return False
|
||
|
||
def normalize(value: str) -> str:
|
||
value = unicodedata.normalize("NFKC", str(value))
|
||
return re.sub(r"\s+", "", value)
|
||
|
||
return normalize(original) == normalize(suggestion)
|
||
|
||
|
||
_HEADING_PREFIX_RE = re.compile(
|
||
r"^\s*(?:"
|
||
r"[一二三四五六七八九十百]+、"
|
||
r"|[（(][一二三四五六七八九十百\d]+[）)]"
|
||
r"|\d+(?:\.\d+)+(?=\s*[\u4e00-\u9fff])"
|
||
r"|\d+(?:\.\d+)*[.、]"
|
||
r")"
|
||
)
|
||
_TRAILING_PUNCTUATION_RE = re.compile(r"[，。；：、,.!?！？:;]+$")
|
||
_FIGURE_CAPTION_RE = re.compile(
|
||
r"^\s*图\s*(?P<number>\d*)\s*[：:]\s*(?P<title>.+?)\s*$"
|
||
)
|
||
|
||
|
||
def _is_standalone_heading_punctuation_change(
|
||
original: str,
|
||
suggestion: str,
|
||
) -> bool:
|
||
"""过滤只修改独占一行编号标题末尾标点的过度校对。"""
|
||
if "\n" in original or "\n" in suggestion:
|
||
return False
|
||
if not _HEADING_PREFIX_RE.match(original):
|
||
return False
|
||
original_body = _TRAILING_PUNCTUATION_RE.sub("", original.strip())
|
||
suggestion_body = _TRAILING_PUNCTUATION_RE.sub("", suggestion.strip())
|
||
return (
|
||
original_body == suggestion_body
|
||
and original.strip() != suggestion.strip()
|
||
)
|
||
|
||
|
||
def _is_figure_numbering_change(
|
||
original: str,
|
||
suggestion: str,
|
||
) -> bool:
|
||
"""
|
||
分块无法获知全文图号。若标题内容未变，只新增、删除或修改图号，
|
||
一律不作为校对结果返回。
|
||
"""
|
||
original_match = _FIGURE_CAPTION_RE.match(original or "")
|
||
suggestion_match = _FIGURE_CAPTION_RE.match(suggestion or "")
|
||
if not original_match or not suggestion_match:
|
||
return False
|
||
return (
|
||
original_match.group("title").strip()
|
||
== suggestion_match.group("title").strip()
|
||
and original_match.group("number")
|
||
!= suggestion_match.group("number")
|
||
)
|
||
|
||
|
||
def _review_content_skeleton(value: str) -> str:
|
||
"""保留字母和数字，用于识别仅标点发生变化的建议。"""
|
||
normalized = unicodedata.normalize("NFKC", str(value))
|
||
return "".join(
|
||
char for char in normalized
|
||
if unicodedata.category(char)[:1] in {"L", "N"}
|
||
)
|
||
|
||
|
||
def _is_punctuation_only_change(original: str, suggestion: str) -> bool:
|
||
return (
|
||
bool(original)
|
||
and bool(suggestion)
|
||
and original != suggestion
|
||
and _review_content_skeleton(original)
|
||
== _review_content_skeleton(suggestion)
|
||
)
|
||
|
||
|
||
def _is_heading_number_change(original: str, suggestion: str) -> bool:
|
||
"""识别标题编号变化，避免模型误标为错别字或逻辑错误。"""
|
||
original_match = _HEADING_PREFIX_RE.match(original)
|
||
suggestion_match = _HEADING_PREFIX_RE.match(suggestion)
|
||
if not original_match or not suggestion_match:
|
||
return False
|
||
return original_match.group() != suggestion_match.group()
|
||
|
||
|
||
def _is_adjacent_repetition_removal(
|
||
original: str,
|
||
suggestion: str,
|
||
) -> bool:
|
||
"""识别“常见常见 -> 常见”这类明确的相邻重复删除。"""
|
||
if not original or len(original) <= len(suggestion):
|
||
return False
|
||
removed_length = len(original) - len(suggestion)
|
||
if removed_length > 50:
|
||
return False
|
||
|
||
for start in range(0, len(original) - removed_length * 2 + 1):
|
||
repeated = original[start:start + removed_length]
|
||
if (
|
||
repeated
|
||
and original[start + removed_length:start + removed_length * 2]
|
||
== repeated
|
||
and original[:start] + original[start + removed_length:]
|
||
== suggestion
|
||
):
|
||
return True
|
||
return False
|
||
|
||
|
||
_CREDENTIAL_EXPOSURE_RE = re.compile(
|
||
r"(?i)(?:"
|
||
r"(?:postgres(?:ql)?|mysql|mongodb(?:\+srv)?|redis)://[^:\s/]+:[^@\s]+@"
|
||
r"|(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*[^\s,，;；]+"
|
||
r")"
|
||
)
|
||
_COMMAND_TEXT_RE = re.compile(
|
||
r"(?i)(?:^|\s)(?:--?[a-z][\w-]*|docker(?:\s|$)|dockker(?:\s|$)|doker(?:\s|$))"
|
||
)
|
||
|
||
|
||
def _contains_credential_exposure(text: str) -> bool:
|
||
return bool(_CREDENTIAL_EXPOSURE_RE.search(text or ""))
|
||
|
||
|
||
def _is_likely_local_text_correction(
|
||
original: str,
|
||
suggestion: str,
|
||
) -> bool:
|
||
"""
|
||
识别错字、漏字、短语润色和命令拼写等局部修改。
|
||
这类修改不能仅通过改写错误标签混入逻辑或合规单选结果。
|
||
"""
|
||
if not original or not suggestion or original == suggestion:
|
||
return False
|
||
if _COMMAND_TEXT_RE.search(original):
|
||
return True
|
||
|
||
edits = [
|
||
(tag, original[i1:i2], suggestion[j1:j2])
|
||
for tag, i1, i2, j1, j2 in SequenceMatcher(
|
||
None,
|
||
original,
|
||
suggestion,
|
||
autojunk=False,
|
||
).get_opcodes()
|
||
if tag != "equal"
|
||
]
|
||
if not edits:
|
||
return False
|
||
|
||
changed_text = "".join(
|
||
original_part + suggestion_part
|
||
for _, original_part, suggestion_part in edits
|
||
)
|
||
if any(char.isdigit() for char in changed_text):
|
||
# 年份、日期、数值的前后冲突可能是真正的逻辑问题。
|
||
return False
|
||
|
||
total_changed = sum(
|
||
len(original_part) + len(suggestion_part)
|
||
for _, original_part, suggestion_part in edits
|
||
)
|
||
max_span = max(
|
||
max(len(original_part), len(suggestion_part))
|
||
for _, original_part, suggestion_part in edits
|
||
)
|
||
return total_changed <= 8 and max_span <= 4
|
||
|
||
|
||
def _minimize_multiline_review_item(
|
||
item: dict,
|
||
source_text: str,
|
||
) -> Optional[dict]:
|
||
"""
|
||
行式返回协议无法承载多行原文。对于“标签换行值”缺少冒号的情况，
|
||
将“名称\\n主发动机 -> 名称：主发动机”缩小为“名称 -> 名称：”。
|
||
其他不能安全缩小的多行修改直接丢弃，避免前端只显示第一行后误替换。
|
||
"""
|
||
original = str(item.get("original", "")).strip()
|
||
suggestion = str(item.get("suggestion", "")).strip()
|
||
if "\n" not in original and "\r" not in original:
|
||
return dict(item)
|
||
|
||
original_lines = [
|
||
line.strip() for line in original.splitlines()
|
||
if line.strip()
|
||
]
|
||
if len(original_lines) < 2 or "\n" in suggestion or "\r" in suggestion:
|
||
return None
|
||
|
||
original_head = original_lines[0]
|
||
unchanged_tail = "".join(original_lines[1:])
|
||
compact_suggestion = re.sub(r"\s+", "", suggestion)
|
||
if not unchanged_tail or not compact_suggestion.endswith(unchanged_tail):
|
||
return None
|
||
|
||
minimized_suggestion = compact_suggestion[:-len(unchanged_tail)]
|
||
if (
|
||
not minimized_suggestion
|
||
or minimized_suggestion == original_head
|
||
or original_head not in source_text
|
||
):
|
||
return None
|
||
|
||
minimized = dict(item)
|
||
minimized["original"] = original_head
|
||
minimized["suggestion"] = minimized_suggestion
|
||
return minimized
|
||
|
||
|
||
def _normalize_review_item_error(item: dict) -> dict:
|
||
"""让明显可判定的错误类型与实际修改保持一致。"""
|
||
normalized = dict(item)
|
||
original = str(normalized.get("original", "")).strip()
|
||
suggestion = str(normalized.get("suggestion", "")).strip()
|
||
|
||
if _contains_credential_exposure(original):
|
||
normalized["error"] = "合规问题"
|
||
elif _is_adjacent_repetition_removal(original, suggestion):
|
||
normalized["error"] = "重复内容"
|
||
elif _is_heading_number_change(original, suggestion):
|
||
normalized["error"] = "格式不规范"
|
||
elif _is_punctuation_only_change(original, suggestion):
|
||
normalized["error"] = "标点不规范"
|
||
elif _is_likely_local_text_correction(original, suggestion):
|
||
normalized["error"] = "错别词语"
|
||
else:
|
||
normalized["error"] = "、".join(
|
||
_split_review_error_types(normalized.get("error", ""))
|
||
)
|
||
return normalized
|
||
|
||
|
||
def repair_review_output(
|
||
model_output: str,
|
||
source_text: str,
|
||
positive_words: Optional[Set[str]] = None,
|
||
sensitive_words: Optional[Set[str]] = None,
|
||
types: Optional[List[str]] = None,
|
||
) -> str:
|
||
"""修复提前改写原文以及修改范围重叠的校对结果。"""
|
||
sensitive_word_set = sensitive_words or set()
|
||
parsed_items = parse_review_output(model_output)
|
||
items = []
|
||
for item in parsed_items:
|
||
minimized_item = _minimize_multiline_review_item(item, source_text)
|
||
if minimized_item is not None:
|
||
normalized_item = _normalize_review_item_error(minimized_item)
|
||
if (
|
||
str(minimized_item.get("original", "")).strip()
|
||
in sensitive_word_set
|
||
and "敏感词汇" in _split_review_error_types(
|
||
minimized_item.get("error", "")
|
||
)
|
||
):
|
||
normalized_item["error"] = "敏感词汇"
|
||
items.append(normalized_item)
|
||
if not items:
|
||
return "###\n###"
|
||
|
||
corrections = [
|
||
item for item in items
|
||
if item.get("original")
|
||
and item.get("suggestion")
|
||
and item["original"] != item["suggestion"]
|
||
and not _is_whitespace_only_change(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)
|
||
|
||
allowed_error_types = (
|
||
get_allowed_review_error_types(types)
|
||
if types is not None
|
||
else None
|
||
)
|
||
if allowed_error_types is not None and not sensitive_word_set:
|
||
allowed_error_types.discard("敏感词汇")
|
||
final_items = []
|
||
seen_records = set()
|
||
for index, item in enumerate(repaired_items):
|
||
original = item.get("original", "").strip()
|
||
suggestion = item.get("suggestion", "").strip()
|
||
error_types = _split_review_error_types(item.get("error", ""))
|
||
suggestion_is_valid = bool(suggestion) or "重复内容" in error_types
|
||
if (
|
||
index in covered_indexes
|
||
or not original
|
||
or not suggestion_is_valid
|
||
or not error_types
|
||
or original not in source_text
|
||
or (
|
||
bool(suggestion)
|
||
and _is_whitespace_only_change(original, suggestion)
|
||
)
|
||
or _is_standalone_heading_punctuation_change(original, suggestion)
|
||
or _is_figure_numbering_change(original, suggestion)
|
||
or _is_heading_number_change(original, suggestion)
|
||
):
|
||
continue
|
||
if (
|
||
allowed_error_types is not None
|
||
and any(
|
||
error_type not in allowed_error_types
|
||
for error_type in error_types
|
||
)
|
||
):
|
||
continue
|
||
if (
|
||
"敏感词汇" in error_types
|
||
and original not in sensitive_word_set
|
||
):
|
||
continue
|
||
normalized_item = dict(item)
|
||
normalized_item["error"] = "、".join(error_types)
|
||
record_key = (
|
||
original,
|
||
normalized_item["error"],
|
||
suggestion,
|
||
)
|
||
if record_key in seen_records:
|
||
continue
|
||
seen_records.add(record_key)
|
||
final_items.append(normalized_item)
|
||
|
||
# 正词命中的“错别字”记录不视为错误，例如“维休”被定义为正词时，
|
||
# 不返回“维休”→“维修”的错别字建议。
|
||
final_items = filter_positive_word_false_positives(
|
||
final_items,
|
||
positive_words or set(),
|
||
)
|
||
|
||
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()
|
||
]
|
||
if not blocks:
|
||
return "###\n###"
|
||
return "###\n" + "\n\n".join(blocks) + "\n###"
|
||
|
||
|
||
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
|
||
if _has_repeating_review_tail(full_text):
|
||
logger.warning("[公文校对] 检测到模型连续重复同一条记录，提前结束本块输出")
|
||
break
|
||
|
||
return full_text
|
||
|
||
|
||
def build_review_stream_require(
|
||
require: Optional[str],
|
||
types: List[str],
|
||
sensitive_words: List[str],
|
||
negative_words: List[str],
|
||
positive_words: List[str],
|
||
) -> Optional[str]:
|
||
requirements = []
|
||
|
||
if require and require.strip():
|
||
requirements.append(require.strip())
|
||
|
||
selected_types = set(normalize_review_types(types))
|
||
|
||
sensitive_text = "、".join(dict.fromkeys(word.strip() for word in sensitive_words if word and word.strip()))
|
||
if sensitive_text and "合规性检查" in selected_types:
|
||
requirements.append(f"请重点检查以下敏感词：{sensitive_text}。如果原文出现这些词，错误类型写“敏感词汇”。")
|
||
|
||
negative_text = "、".join(dict.fromkeys(word.strip() for word in negative_words if word and word.strip()))
|
||
if negative_text and "基础校对" in selected_types:
|
||
requirements.append(f"请重点检查以下错误词：{negative_text}。如果原文出现这些词，错误类型写“错误词汇”。")
|
||
|
||
positive_text = "、".join(dict.fromkeys(word.strip() for word in positive_words if word and word.strip()))
|
||
if positive_text and "基础校对" in selected_types:
|
||
requirements.append(f"以下词为正确词，涉及错别字判断时不要误判：{positive_text}。")
|
||
|
||
if not requirements:
|
||
return None
|
||
|
||
return "\n".join(requirements)
|
||
|
||
|
||
|
||
def split_review_chunks(content: str, max_size: int = REVIEW_CHUNK_SIZE) -> List[str]:
|
||
"""优先按段落、其次按句末标点切分，单块不超过 max_size。"""
|
||
if not content:
|
||
return []
|
||
if max_size <= 0:
|
||
raise ValueError("max_size must be greater than zero")
|
||
|
||
chunks: List[str] = []
|
||
start = 0
|
||
while start < len(content):
|
||
if len(content) - start <= max_size:
|
||
chunks.append(content[start:])
|
||
break
|
||
|
||
candidate = content[start:start + max_size]
|
||
paragraph_matches = list(re.finditer(r"\n\s*", candidate))
|
||
if paragraph_matches:
|
||
cut = paragraph_matches[-1].end()
|
||
else:
|
||
sentence_matches = list(re.finditer(
|
||
r"[。！？；.!?;](?:[\"'”’）)】》])?\s*", candidate))
|
||
cut = sentence_matches[-1].end() if sentence_matches else max_size
|
||
|
||
chunks.append(content[start:start + cut])
|
||
start += cut
|
||
|
||
return chunks
|
||
|
||
|
||
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,
|
||
types=types,
|
||
sensitive_words=sensitive_words,
|
||
negative_words=negative_words,
|
||
positive_words=positive_words,
|
||
)
|
||
|
||
chunks = split_review_chunks(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)
|
||
repaired_output = repair_review_output(
|
||
model_output,
|
||
chunk,
|
||
positive_words=normalize_words(positive_words),
|
||
sensitive_words=normalize_words(sensitive_words),
|
||
types=types,
|
||
)
|
||
# 仅打印过滤后的最终保留记录，不打印模型原始输出或中间结果。
|
||
if repaired_output.strip() and repaired_output.strip() != "###\n###":
|
||
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)
|
||
|
||
chunks = split_review_chunks(content, REVIEW_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(
|
||
item for item in llm_items
|
||
if item.get("original", "").strip()
|
||
and item.get("error", "").strip()
|
||
and not _is_whitespace_only_change(
|
||
item.get("original", ""), item.get("suggestion", "")
|
||
)
|
||
)
|
||
|
||
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 _citation_display_filename(value: Any) -> Any:
|
||
"""仅转换文本溯源展示名，不修改 file_id 或实际文件路径。"""
|
||
if not isinstance(value, str):
|
||
return value
|
||
return re.sub(
|
||
r"(?i)\.(?:pptx|ppt)$",
|
||
".pdf",
|
||
value.strip(),
|
||
)
|
||
|
||
|
||
def normalize_source_citation_filenames(
|
||
source_citation: Dict[str, Any],
|
||
) -> Dict[str, Any]:
|
||
"""将最终文本溯源中的 PPT/PPTX 展示后缀统一转换为 PDF。"""
|
||
if not isinstance(source_citation, dict):
|
||
return source_citation
|
||
|
||
normalized: Dict[str, Any] = {}
|
||
for resource_name, items in source_citation.items():
|
||
display_resource = _citation_display_filename(resource_name)
|
||
|
||
if isinstance(items, list):
|
||
normalized_items = []
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
normalized_items.append(item)
|
||
continue
|
||
normalized_item = dict(item)
|
||
for field in ("filename", "resource", "title"):
|
||
if field in normalized_item:
|
||
normalized_item[field] = _citation_display_filename(
|
||
normalized_item[field]
|
||
)
|
||
normalized_items.append(normalized_item)
|
||
else:
|
||
normalized_items = items
|
||
|
||
# 极少数同名 .ppt/.pptx 会映射到同一 .pdf，保留全部切片。
|
||
if (
|
||
display_resource in normalized
|
||
and isinstance(normalized[display_resource], list)
|
||
and isinstance(normalized_items, list)
|
||
):
|
||
normalized[display_resource].extend(normalized_items)
|
||
else:
|
||
normalized[display_resource] = normalized_items
|
||
|
||
return normalized
|
||
|
||
|
||
|
||
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(
|
||
normalize_source_citation_filenames(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
|
||
has_streamed_content = 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)}")
|
||
# 装饰器产生的无标题事件属于内部入参/返回值，不应在前端展示。
|
||
# 前端会把它们自动编号成“步骤 2”“步骤 5”，从而泄露检索原文
|
||
# 和未整理的回答草稿。
|
||
if not title.strip():
|
||
continue
|
||
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", "")
|
||
|
||
if content:
|
||
has_streamed_content = True
|
||
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 "")
|
||
frontend_final_result = final_result
|
||
if has_streamed_content and final_result:
|
||
# 正文已按增量流式发送，完成事件只补充 actions 等元数据。
|
||
# 再携带完整 content 会被前端追加为第二份重复正文。
|
||
frontend_final_result = {**final_result, "content": ""}
|
||
yield format_stream_event("execution_complete", {
|
||
"final_result": frontend_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.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]
|
||
|
||
|
||
class Base64ImageRequest(BaseModel):
|
||
content: Optional[str] = Field(
|
||
None,
|
||
description="Optional text around the image, for example: 图3.1 船舶主发动机组成图（images/test.jpg）",
|
||
)
|
||
filename: str = Field(..., description="Original image filename, for example test.png")
|
||
image_base64: str = Field(..., description="Base64 image content, with or without data:image/... prefix")
|
||
|
||
|
||
async def analyze_image_with_vlm(image_bytes: bytes, suffix: str) -> str:
|
||
image_url = image_base64_to_data_url(image_bytes, suffix)
|
||
async with VLM_SEMAPHORE:
|
||
return await model_api.OpenaiAPI.open_api_vl_without_thinking(image_url)
|
||
|
||
|
||
@app.post("/split_image")
|
||
async def split_image(request_data: Base64ImageRequest):
|
||
filename = safe_original_filename(request_data.filename)
|
||
suffix = PathLib(filename).suffix.lower()
|
||
if suffix not in {".jpg", ".jpeg", ".png", ".bmp", ".webp", ".gif", ".tif", ".tiff"}:
|
||
raise HTTPException(status_code=400, detail="Unsupported image type")
|
||
|
||
image_bytes, processed_suffix = prepare_split_image(request_data.image_base64, suffix)
|
||
if not image_bytes:
|
||
raise HTTPException(status_code=400, detail="decoded image is empty")
|
||
|
||
temp_path = None
|
||
try:
|
||
with tempfile.NamedTemporaryFile(delete=False, suffix=processed_suffix) as tmp:
|
||
tmp.write(image_bytes)
|
||
temp_path = PathLib(tmp.name)
|
||
|
||
ocr_data = await extract_image_text(temp_path)
|
||
try:
|
||
temp_path.unlink()
|
||
except Exception as e:
|
||
logger.warning(f"Failed to delete temp image {temp_path}: {e}")
|
||
finally:
|
||
temp_path = None
|
||
|
||
content = (request_data.content or "").strip()
|
||
ocr_full_text = (ocr_data.get("full_text") or "").strip()
|
||
|
||
if not content and not ocr_full_text:
|
||
vlm_text = await analyze_image_with_vlm(image_bytes, processed_suffix)
|
||
image_content = "图片文本描述为：" + (vlm_text or "").strip()
|
||
elif content and ocr_full_text:
|
||
image_content = f"图片上下文内容为：{content}, {ocr_full_text}"
|
||
elif content:
|
||
image_content = "图片上下文内容为：" + content
|
||
else:
|
||
image_content = ocr_full_text
|
||
|
||
image_content = "图片文件名为：" + filename + "，图片相关内容为：" + image_content
|
||
return JSONResponse(
|
||
{
|
||
"code": 200,
|
||
"message": "ok",
|
||
"data": {
|
||
"image_content": image_content,
|
||
"filename": filename,
|
||
},
|
||
}
|
||
)
|
||
finally:
|
||
if temp_path and temp_path.exists():
|
||
try:
|
||
temp_path.unlink()
|
||
except Exception as e:
|
||
logger.warning(f"Failed to delete temp image {temp_path}: {e}")
|
||
|
||
|
||
|
||
|
||
@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}")
|
||
|
||
# 执行转换
|
||
get_doc2pdf_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("/api/health")
|
||
async def health_check():
|
||
"""健康检查接口"""
|
||
return {
|
||
"status": "healthy",
|
||
"message": "服务正常运行，每次请求时创建新的Agent实例"
|
||
}
|
||
|
||
@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)
|