1704 lines
65 KiB
Python
1704 lines
65 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())
|
||
|
||
import ahocorasick
|
||
from pydantic import BaseModel, Field, model_validator
|
||
from fastapi import FastAPI, File, Path, UploadFile, HTTPException, Form, Request, Header, Body, BackgroundTasks
|
||
from fastapi.responses import StreamingResponse
|
||
try:
|
||
from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate
|
||
except ImportError:
|
||
from langchain.prompts import FewShotPromptTemplate, PromptTemplate
|
||
from openai.types.chat import ChatCompletionSystemMessageParam, ChatCompletionUserMessageParam, ChatCompletionMessageParam
|
||
from chunk_text import get_chunk_bbox, split_text_preserve_sentences, data_replace, chunk_check,merge_short_slices,find_and_read_content_list,process_pdf_file,process_other_file
|
||
from fastapi.responses import JSONResponse, StreamingResponse
|
||
import requests
|
||
from pathlib import PurePath, Path as PathLib
|
||
import tempfile
|
||
|
||
from modelsAPI import model_api
|
||
from documents_prompt import (
|
||
BEAUTIFY_TYPE_DICT,
|
||
TEMPLATE_DICT,
|
||
GENERATE_NORMAL_OUTLINE_PROMPT,
|
||
GENERATE_MARKDOWN_LIST_OUTLINE_PROMPT,
|
||
GENERATE_METHOD_OUTLINE_PROMPT,
|
||
)
|
||
"""app.py
|
||
FastAPI 主 Agent 接口 - 修复重复执行问题
|
||
提供主脑 Agent 的 HTTP API 服务，支持流式输出
|
||
集成 LangGraph Checkpointer 实现状态持久化
|
||
"""
|
||
import logging
|
||
from time import sleep
|
||
import aiofiles
|
||
|
||
from dotenv import load_dotenv
|
||
from fastapi import FastAPI, HTTPException
|
||
from fastapi.responses import StreamingResponse, FileResponse
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from fastapi.staticfiles import StaticFiles
|
||
from pydantic import BaseModel
|
||
from typing import Optional, Dict, Any, AsyncGenerator,List
|
||
import uvicorn
|
||
import json
|
||
import os
|
||
import uuid
|
||
from main_agent import create_main_agent
|
||
from utils.function_tracker import (
|
||
register_event_callback,
|
||
create_stream_event_handler,
|
||
unregister_event_callback,
|
||
clear_current_stream_handler
|
||
)
|
||
import asyncio
|
||
import importlib
|
||
from doc2pdf import Doc2PDF
|
||
|
||
from workflow_registry import WORKFLOW_CONFIG, VALID_ROUTE_FLAGS
|
||
from main_agent import extract_conversation_title
|
||
from workflows.history_manager import filter_image_urls, preprocess_from_request
|
||
from config import set_request_user_config
|
||
from checkpointer_config import (
|
||
CheckpointerManager,
|
||
checkpointer_manager
|
||
)
|
||
app = FastAPI(max_request_size=1024 * 1024 * 10)
|
||
|
||
converter = Doc2PDF()
|
||
DATA_DIR = "/app/files"
|
||
load_dotenv()
|
||
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger(__name__)
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
# app.mount("/picture", StaticFiles(directory="/app/picture"), name="pictures")
|
||
|
||
STREAM_RESPONSE_HEADERS = {
|
||
"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no",
|
||
"X-Content-Type-Options": "nosniff",
|
||
}
|
||
|
||
|
||
def llm_text_stream_response(content) -> StreamingResponse:
|
||
return StreamingResponse(
|
||
content,
|
||
media_type="text/plain; charset=utf-8",
|
||
headers=STREAM_RESPONSE_HEADERS,
|
||
)
|
||
|
||
|
||
@app.on_event("startup")
|
||
async def startup_event():
|
||
"""应用启动时初始化"""
|
||
await checkpointer_manager.setup()
|
||
# 初始化智能体使用统计表
|
||
from tools.agent_usage_statistics import init_agent_usage_table
|
||
await init_agent_usage_table()
|
||
logger.info("应用启动完成，PostgreSQL Checkpointer 已配置")
|
||
|
||
|
||
# =============== PDF 文件目录配置 ===============
|
||
# 创建 PDF 文件目录（用于存储生成的 PDF 等文件）
|
||
PDF_DIR = os.path.join(os.path.dirname(__file__), "created_pdf")
|
||
# 创建 word 文件目录（用于存储生成的 word 等文件）
|
||
WORD_DIR = os.path.join(os.path.dirname(__file__), "created_word")
|
||
os.makedirs(PDF_DIR, exist_ok=True)
|
||
|
||
class RewriteRequest(BaseModel):
|
||
rewrite_type: Optional[str] = None
|
||
content: Optional[str] = None
|
||
require: Optional[str] = None
|
||
sentence: Optional[str] = None
|
||
|
||
|
||
REWRITE_TYPE_MAP = {
|
||
"quotePolish": "金句润色",
|
||
"formalStyle": "书面化",
|
||
"speechStyle": "讲稿化",
|
||
"formatting": "规整",
|
||
"expand": "扩写",
|
||
"continueWriting": "续写",
|
||
"report": "汇报",
|
||
"quote": "金句",
|
||
"summarize": "精简",
|
||
"condense": "总结",
|
||
}
|
||
|
||
|
||
def build_rewrite_prompt(request: RewriteRequest) -> str:
|
||
beautify_type = REWRITE_TYPE_MAP.get(request.rewrite_type or "")
|
||
if not beautify_type:
|
||
raise HTTPException(status_code=400, detail=f"不支持的 rewrite_type: {request.rewrite_type}")
|
||
|
||
prompt_template = BEAUTIFY_TYPE_DICT.get(beautify_type)
|
||
if not prompt_template:
|
||
raise HTTPException(status_code=400, detail=f"未找到提示词模板: {beautify_type}")
|
||
|
||
return prompt_template.format(
|
||
require=request.require or "",
|
||
content=request.content or "",
|
||
sentence=request.sentence or request.content or "",
|
||
)
|
||
|
||
|
||
async def stream_llm_content(prompt: str):
|
||
async for chunk in model_api.OpenaiAPI.open_api_chat_stream(
|
||
query=prompt,
|
||
model=None,
|
||
system_prompt="你是专业文稿改写助手，根据要求进行文本改写。",
|
||
messages=[],
|
||
):
|
||
if chunk:
|
||
yield chunk
|
||
|
||
|
||
@app.post("/Rewrite")
|
||
async def rewrite(request: RewriteRequest):
|
||
prompt = build_rewrite_prompt(request)
|
||
|
||
return llm_text_stream_response(stream_llm_content(prompt))
|
||
|
||
|
||
class WriteRequest(BaseModel):
|
||
role: Optional[str] = None
|
||
template: Optional[str] = None
|
||
title: Optional[str] = None
|
||
length: Optional[str] = None
|
||
requirement: Optional[str] = None
|
||
references: Optional[str] = None
|
||
outline: Optional[str] = None
|
||
konwlege: Optional[str] = None
|
||
prompt: Optional[str] = None
|
||
|
||
class OutlineRequest(BaseModel):
|
||
role: Optional[str] = None
|
||
template: Optional[str] = None
|
||
title: Optional[str] = None
|
||
length: Optional[str] = None
|
||
requirement: Optional[str] = None
|
||
references: Optional[str] = None
|
||
konwlege: Optional[str] = None
|
||
prompt: Optional[str] = None
|
||
|
||
|
||
def length_convert(length: Optional[str]):
|
||
"""将篇幅解析为 {target, max}：target 为建议目标字数，max 为硬上限。"""
|
||
if length == "短":
|
||
return {"target": 700, "max": 1000}
|
||
if length == "中":
|
||
return {"target": 2500, "max": 3000}
|
||
if length == "长":
|
||
return {"target": 4500, "max": 5000}
|
||
# 兜底：从自定义篇幅字符串里解析出最大字数
|
||
if length:
|
||
nums = re.findall(r"\d+", length)
|
||
if nums:
|
||
max_chars = int(nums[-1])
|
||
return {"target": int(max_chars * 0.8), "max": max_chars}
|
||
return None
|
||
|
||
|
||
def length_display(length: Optional[str]) -> str:
|
||
"""模板 {length} 占位符用的字符串。"""
|
||
info = length_convert(length)
|
||
if info:
|
||
return f"{info['target']}字左右（不超过{info['max']}字）"
|
||
return length or ""
|
||
|
||
|
||
def build_length_instruction(length: Optional[str]) -> str:
|
||
"""把 target/max 写进提示词：建议目标 + 硬上限 + 优先级。"""
|
||
info = length_convert(length)
|
||
if not info:
|
||
return ""
|
||
return (
|
||
f"全文建议写到{info['target']}字左右，绝对不得超过{info['max']}字；"
|
||
"字数为硬性要求，不得注水或重复凑字数。"
|
||
)
|
||
|
||
|
||
def build_write_prompt(request: WriteRequest) -> str:
|
||
prompt = request.prompt or ""
|
||
template = request.template or ""
|
||
length_instruction = build_length_instruction(request.length)
|
||
content = f"请撰写一篇{template}。{length_instruction}{prompt}"
|
||
prompt_template = TEMPLATE_DICT.get(template)
|
||
if not prompt_template:
|
||
raise HTTPException(status_code=400, detail=f"不支持的写作模板: {template}")
|
||
|
||
title = request.title or ""
|
||
if not title.strip():
|
||
raise HTTPException(status_code=400, detail="title不能为空")
|
||
|
||
return 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 "",
|
||
)
|
||
|
||
|
||
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 llm_text_stream_response(stream_write_content(prompt))
|
||
|
||
|
||
def build_outline_prompt(request: OutlineRequest) -> str:
|
||
if request.prompt and request.prompt.strip():
|
||
return request.prompt.strip()
|
||
|
||
template = request.template or ""
|
||
title = request.title or ""
|
||
if not title.strip():
|
||
raise HTTPException(status_code=400, detail="title不能为空")
|
||
|
||
markdown_list_template = {"报告", "函", "文书", "工作总结"}
|
||
prompt_template = GENERATE_NORMAL_OUTLINE_PROMPT
|
||
|
||
if template == "方案":
|
||
prompt_template = GENERATE_METHOD_OUTLINE_PROMPT
|
||
elif template in markdown_list_template:
|
||
prompt_template = GENERATE_MARKDOWN_LIST_OUTLINE_PROMPT
|
||
|
||
return prompt_template.format(
|
||
role=request.role or "",
|
||
template=template,
|
||
title=title,
|
||
length=length_display(request.length),
|
||
requirement=request.requirement or "",
|
||
references=request.references or "",
|
||
)
|
||
|
||
|
||
async def stream_outline_content(prompt: str):
|
||
async for chunk in model_api.OpenaiAPI.open_api_chat_stream(
|
||
query=prompt,
|
||
model=None,
|
||
system_prompt="你是专业公文提纲生成助手，请直接输出提纲内容。",
|
||
messages=[],
|
||
):
|
||
if chunk:
|
||
yield chunk
|
||
|
||
|
||
@app.post("/Outline")
|
||
async def outline(request: OutlineRequest):
|
||
prompt = build_outline_prompt(request)
|
||
|
||
return llm_text_stream_response(stream_outline_content(prompt))
|
||
|
||
|
||
class ReviewRequest(BaseModel):
|
||
require: Optional[str] = Field(None, description="额外校对要求")
|
||
content: str = Field(..., description="待校对文本")
|
||
types: List[str] = Field(..., description="检查类型")
|
||
sensitive_words: List[str] = Field(default_factory=list, description="敏感词汇")
|
||
negative_words: List[str] = Field(default_factory=list, description="错误词汇")
|
||
positive_words: List[str] = Field(default_factory=list, description="正词词汇")
|
||
|
||
@model_validator(mode="after")
|
||
def check_content(self):
|
||
if not self.content or not self.content.strip():
|
||
raise ValueError("content不能为空")
|
||
return self
|
||
|
||
|
||
class ReviewItem(BaseModel):
|
||
original: str
|
||
error: str
|
||
suggestion: str
|
||
|
||
|
||
class ReviewResponse(BaseModel):
|
||
data: List[ReviewItem]
|
||
|
||
|
||
def normalize_words(words: List[str]) -> Set[str]:
|
||
return {word.strip() for word in words if word and word.strip()}
|
||
|
||
|
||
def find_keywords(text: str, keyword_set: Set[str]) -> List[str]:
|
||
if not text or not keyword_set:
|
||
return []
|
||
|
||
automaton = ahocorasick.Automaton()
|
||
|
||
for idx, keyword in enumerate(keyword_set):
|
||
automaton.add_word(keyword, (idx, keyword))
|
||
|
||
automaton.make_automaton()
|
||
|
||
found = set()
|
||
for _, (_, keyword) in automaton.iter(text):
|
||
found.add(keyword)
|
||
|
||
return list(found)
|
||
|
||
|
||
def append_dictionary_results(
|
||
items: List[dict],
|
||
content: str,
|
||
sensitive_words: Set[str],
|
||
negative_words: Set[str],
|
||
) -> List[dict]:
|
||
for word in find_keywords(content, sensitive_words):
|
||
items.append({
|
||
"original": word,
|
||
"error": "敏感词汇",
|
||
"suggestion": "",
|
||
})
|
||
|
||
for word in find_keywords(content, negative_words):
|
||
items.append({
|
||
"original": word,
|
||
"error": "错误词汇",
|
||
"suggestion": "",
|
||
})
|
||
|
||
return items
|
||
|
||
|
||
def filter_positive_word_false_positives(
|
||
items: List[dict],
|
||
positive_words: Set[str],
|
||
) -> List[dict]:
|
||
if not positive_words:
|
||
return items
|
||
|
||
result = []
|
||
|
||
for item in items:
|
||
original = str(item.get("original", ""))
|
||
error = str(item.get("error", ""))
|
||
|
||
has_positive_word = any(word in original for word in positive_words)
|
||
is_typo_error = (
|
||
"错别字" in error
|
||
or "错别词" in error
|
||
or "错别词语" in error
|
||
)
|
||
|
||
if has_positive_word and is_typo_error:
|
||
continue
|
||
|
||
result.append(item)
|
||
|
||
return result
|
||
|
||
|
||
def dedupe_and_filter(items: List[dict], source_text: str) -> List[dict]:
|
||
seen = set()
|
||
result = []
|
||
|
||
for item in items:
|
||
original = str(item.get("original", "")).strip()
|
||
error = str(item.get("error", "")).strip()
|
||
suggestion = str(item.get("suggestion", "")).strip()
|
||
|
||
if not original:
|
||
continue
|
||
|
||
if original == suggestion:
|
||
continue
|
||
|
||
if original not in source_text:
|
||
continue
|
||
|
||
key = (original, error, suggestion)
|
||
if key in seen:
|
||
continue
|
||
|
||
seen.add(key)
|
||
result.append({
|
||
"original": original,
|
||
"error": error,
|
||
"suggestion": suggestion,
|
||
})
|
||
|
||
return result
|
||
|
||
def cotprompt(query):
|
||
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 = ""
|
||
require = require.strip()
|
||
review_t = "现在需要你帮我完成以下文本校对任务："
|
||
for i , t in enumerate(types):
|
||
if t =="逻辑校对":
|
||
review_t += f"{i} 对这段文本进行逻辑校对\n"
|
||
if t =="基础校对":
|
||
review_t += f"{i}.1 进行错别字校对，错别字错误主要以文本用字不当为主，例如：星光店电，正确的应该为：星光点点。\n"
|
||
review_t += f"{i}.2 进行标点校对，主要以标点符号使用不当为主，例如：星光点点；万里无云，正确的应该为：星光点点，万里无云\n"
|
||
review_t += f"{i}.3 进行格式规范校对\n"
|
||
review_t += f"{i}.4 进行重复内容校对，主要以原文中出现重复词语、句子或文本为主，例如：“今天今天心情相当不错，我很开心。我很开心。“，正确的内容应该为：今天心情相当不错，我很开心。\n"
|
||
if t=="合规性检查":
|
||
review_t += f"{i} 对这段文本进行合规性检查\n"
|
||
require_text = f"\n额外要求：{require}\n" if require else ""
|
||
query = f"""
|
||
我会给你对应的文本，进行相应的检查，{review_t}
|
||
{require_text}
|
||
"""
|
||
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<error>.*?)\s*[\r\n]+"
|
||
r"原文[:：]\s*(?P<original>.*?)\s*[\r\n]+"
|
||
r"建议[:：]\s*(?P<suggestion>.*?)(?=\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
|
||
|
||
|
||
REVIEW_SYSTEM_PROMPT = (
|
||
"你是一个专业的文本校对专家，能够根据要求进行内容的纠正和改错"
|
||
"请严格按照用户要求的格式输出。"
|
||
"不要输出推理过程、解释性文字、前缀或总结。"
|
||
)
|
||
|
||
|
||
async def stream_review_model(prompt: Iterable[ChatCompletionMessageParam]):
|
||
async for chunk in model_api.OpenaiAPI.open_api_chat_stream(
|
||
model=None,
|
||
json_output=False,
|
||
system_prompt=REVIEW_SYSTEM_PROMPT,
|
||
messages=prompt,
|
||
enable_thinking=False,
|
||
temperature=0.1,
|
||
):
|
||
if chunk:
|
||
yield chunk
|
||
|
||
|
||
async def call_review_model(prompt: Iterable[ChatCompletionMessageParam]) -> str:
|
||
full_text = ""
|
||
|
||
async for chunk in stream_review_model(prompt):
|
||
full_text += chunk
|
||
|
||
return full_text
|
||
|
||
|
||
def build_review_stream_require(
|
||
require: Optional[str],
|
||
sensitive_words: List[str],
|
||
negative_words: List[str],
|
||
positive_words: List[str],
|
||
) -> Optional[str]:
|
||
requirements = []
|
||
|
||
if require and require.strip():
|
||
requirements.append(require.strip())
|
||
|
||
sensitive_text = "、".join(dict.fromkeys(word.strip() for word in sensitive_words if word and word.strip()))
|
||
if sensitive_text:
|
||
requirements.append(f"请重点检查以下敏感词：{sensitive_text}。如果原文出现这些词，错误类型写“敏感词汇”。")
|
||
|
||
negative_text = "、".join(dict.fromkeys(word.strip() for word in negative_words if word and word.strip()))
|
||
if negative_text:
|
||
requirements.append(f"请重点检查以下错误词：{negative_text}。如果原文出现这些词，错误类型写“错误词汇”。")
|
||
|
||
positive_text = "、".join(dict.fromkeys(word.strip() for word in positive_words if word and word.strip()))
|
||
if positive_text:
|
||
requirements.append(f"以下词为正确词，涉及错别字判断时不要误判：{positive_text}。")
|
||
|
||
if not requirements:
|
||
return None
|
||
|
||
return "\n".join(requirements)
|
||
|
||
|
||
async def stream_review_content(
|
||
content: str,
|
||
types: List[str],
|
||
require: Optional[str],
|
||
sensitive_words: List[str],
|
||
negative_words: List[str],
|
||
positive_words: List[str],
|
||
):
|
||
review_require = build_review_stream_require(
|
||
require=require,
|
||
sensitive_words=sensitive_words,
|
||
negative_words=negative_words,
|
||
positive_words=positive_words,
|
||
)
|
||
|
||
chunk_size = 1000
|
||
chunks = [
|
||
content[i:i + chunk_size]
|
||
for i in range(0, len(content), chunk_size)
|
||
]
|
||
|
||
for chunk in chunks:
|
||
prompt = build_review_prompt(
|
||
content=chunk,
|
||
types=types,
|
||
require=review_require,
|
||
)
|
||
|
||
async for token in stream_review_model(prompt):
|
||
yield token
|
||
|
||
|
||
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")
|
||
async def review(request: ReviewRequest):
|
||
return llm_text_stream_response(stream_review_content(
|
||
content=request.content,
|
||
types=request.types,
|
||
require=request.require,
|
||
sensitive_words=request.sensitive_words,
|
||
negative_words=request.negative_words,
|
||
positive_words=request.positive_words,
|
||
))
|
||
|
||
|
||
|
||
# =============== 请求模型 ===============
|
||
|
||
|
||
class AgentRequest(BaseModel):
|
||
"""Agent 请求模型"""
|
||
query: Optional[Any] = None # 用户文本输入
|
||
file_text: Optional[str] = None # 文件文本输入
|
||
image: Optional[str] = None # 图片路径
|
||
audio: Optional[str] = None # 音频路径（支持 audio 或 asr）
|
||
|
||
# 路由标志（可选，如果提供则跳过任务分类）
|
||
route_flag: Optional[str] = None # 路由标志
|
||
history_message: Optional[Any] = None # 历史信息（支持字符串或列表格式）
|
||
|
||
# 接收参数
|
||
top_k: Optional[Any] = None # rag 前top_k
|
||
rag_prompt: Optional[Any] = None # rag 提示词模板（支持字符串、列表、字典等多种类型）
|
||
image_kb_id: Optional[str] = None # 图库 知识库id
|
||
image_file_id: Optional[str] = None # 图库 文件id
|
||
text_kb_id: Optional[str] = None # 检索 知识库id
|
||
text_file_id: Optional[str] = None # 检索 文件id
|
||
|
||
# Socket.IO 格式所需字段
|
||
chat_id: Optional[str] = None # 聊天ID
|
||
message_id: Optional[str] = None # 消息ID
|
||
|
||
# 知识库配置
|
||
x_user_id: Optional[str] = "1" # 用户ID
|
||
x_user_name: Optional[str] = "testuser" # 用户名称
|
||
x_role: Optional[str] = "admin" # 角色
|
||
|
||
|
||
class AgentResponse(BaseModel):
|
||
"""Agent 响应模型"""
|
||
success: bool
|
||
message: str
|
||
data: Optional[Dict[str, Any]] = None
|
||
error: Optional[str] = None
|
||
|
||
class FeedbackClassifyRequest(BaseModel):
|
||
"""反馈分类请求模型"""
|
||
feedback_text: str
|
||
|
||
|
||
|
||
# =============== 辅助函数 ===============
|
||
def process_rag_prompt(rag_prompt: Any) -> str:
|
||
"""
|
||
处理 rag_prompt 参数，将各种类型转换为字符串
|
||
|
||
Args:
|
||
rag_prompt: 可以是字符串、列表、字典等任意类型
|
||
|
||
Returns:
|
||
处理后的字符串
|
||
"""
|
||
if rag_prompt is None:
|
||
return ""
|
||
|
||
# 如果是字符串，直接返回
|
||
if isinstance(rag_prompt, str):
|
||
return rag_prompt.strip()
|
||
|
||
# 如果是列表，转换为多行字符串
|
||
if isinstance(rag_prompt, list):
|
||
# 将列表中的每个元素转换为字符串，并用换行符连接
|
||
return "\n".join(str(item) for item in rag_prompt if item)
|
||
|
||
# 如果是字典，转换为 JSON 字符串
|
||
if isinstance(rag_prompt, dict):
|
||
try:
|
||
return json.dumps(rag_prompt, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
logger.warning(f"rag_prompt 字典转换失败: {e}")
|
||
return str(rag_prompt)
|
||
|
||
# 其他类型，直接转换为字符串
|
||
return str(rag_prompt).strip()
|
||
|
||
|
||
def format_stream_event(event_type: str,
|
||
data: Dict[str, Any],
|
||
chat_id: str = "",
|
||
message_id: str = "",
|
||
type: str = "chat:completion",
|
||
) -> str:
|
||
"""格式化流式事件输出为 Socket.IO 格式"""
|
||
# 生成 message_id（如果没有提供）
|
||
if not message_id:
|
||
message_id = str(uuid.uuid4())
|
||
|
||
# 根据事件类型构建不同的数据结构
|
||
if event_type == "stream_content":
|
||
# stream_content 事件使用 chat:completion 格式
|
||
content = data.get("content", "")
|
||
event_data = {
|
||
"chat_id": chat_id,
|
||
"message_id": message_id,
|
||
"data": {
|
||
"type": "chat:completion",
|
||
"data": {
|
||
"content": content
|
||
}
|
||
}
|
||
}
|
||
elif event_type == "function_end":
|
||
# function_end 事件
|
||
event_data = {
|
||
"chat_id": chat_id,
|
||
"message_id": message_id,
|
||
"data": {
|
||
"type": "chat:completion",
|
||
"data": {
|
||
"stepsBlueprint": data
|
||
}
|
||
}
|
||
}
|
||
elif event_type == "execution_complete":
|
||
# execution_complete 事件
|
||
|
||
rag_result = data.get("rag_result", {})
|
||
graph_result = data.get("graph_result", [])
|
||
final_result = data.get("final_result", {})
|
||
title = data.get("title", "")
|
||
|
||
# 判断是否有有效的 RAG 结果（非空 dict）
|
||
has_rag = isinstance(rag_result, dict) and bool(rag_result)
|
||
|
||
# 判断是否有有效的图谱结果（非空 list，且至少一个元素的 nodes 非空）
|
||
has_graph = (
|
||
isinstance(graph_result, list)
|
||
and len(graph_result) > 0
|
||
and any(isinstance(item, dict) and item.get("nodes") for item in graph_result)
|
||
)
|
||
|
||
|
||
# 只有当 RAG 或图谱有有效内容时，才构建 sourceCitation
|
||
source_citation = None
|
||
if has_rag or has_graph:
|
||
source_citation = {}
|
||
if has_rag:
|
||
source_citation.update(rag_result)
|
||
if has_graph:
|
||
source_citation["xxxx.graph"] = graph_result
|
||
|
||
# 构建 event_data（title 由调用方异步计算后传入）
|
||
if final_result:
|
||
content = final_result.get("content", "")
|
||
event_data = {
|
||
"chat_id": chat_id,
|
||
"message_id": message_id,
|
||
"data": {
|
||
"type": "chat:completion",
|
||
"data": {
|
||
"content": content,
|
||
},
|
||
"actions": final_result.get("actions", ""),
|
||
"suggestedReplies": final_result.get("suggestedReplies", "")
|
||
}
|
||
}
|
||
else:
|
||
event_data = {
|
||
"chat_id": chat_id,
|
||
"message_id": message_id,
|
||
"data": {
|
||
"type": "chat:completion",
|
||
"data": {
|
||
"data": {},
|
||
"done": True,
|
||
"title": title or "新对话",
|
||
}
|
||
}
|
||
}
|
||
|
||
# 仅在有引用来源时添加 sourceCitation 字段
|
||
if source_citation is not None:
|
||
event_data["data"]["data"]["sourceCitation"] = source_citation
|
||
|
||
else:
|
||
# 其他事件
|
||
event_data = {
|
||
"chat_id": chat_id,
|
||
"message_id": message_id,
|
||
"type": event_type,
|
||
"data": {
|
||
"type": event_type,
|
||
"data": ""
|
||
}
|
||
}
|
||
|
||
# Socket.IO 格式：["chat-events", {...}]
|
||
socket_io_message = ["chat-events", event_data]
|
||
return f"{json.dumps(socket_io_message, ensure_ascii=False)}\n"
|
||
|
||
|
||
|
||
|
||
async def execute_workflow_chain(
|
||
route_flag: str,
|
||
route_params: Dict[str, Any],
|
||
chat_id: str = "",
|
||
message_id: str = "",
|
||
checkpointer=None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
执行工作流
|
||
|
||
Args:
|
||
route_flag: 路由标志
|
||
route_params: 路由参数
|
||
chat_id: 聊天会话 ID
|
||
message_id: 消息 ID
|
||
checkpointer: LangGraph checkpointer 实例
|
||
|
||
Returns:
|
||
工作流执行结果
|
||
"""
|
||
|
||
print("子工作流标志：", route_flag)
|
||
print("子工作流参数：", route_params)
|
||
|
||
if not route_flag:
|
||
return {}
|
||
|
||
extracted_text = route_params.get("extracted_text", "")
|
||
combined_query = route_params.get("combined_query", "")
|
||
history_message = route_params.get("history_message", "")
|
||
|
||
workflow_info = WORKFLOW_CONFIG.get(route_flag)
|
||
|
||
if not workflow_info:
|
||
logger.warning(f"未知的路由标志: {route_flag}，使用默认的qa工作流")
|
||
workflow_info = WORKFLOW_CONFIG.get("qa", WORKFLOW_CONFIG.get(list(WORKFLOW_CONFIG.keys())[0]))
|
||
|
||
try:
|
||
module = importlib.import_module(workflow_info["module"])
|
||
workflow_func = getattr(module, workflow_info["function"])
|
||
except (ImportError, AttributeError) as e:
|
||
print(e)
|
||
logger.error(f"无法导入工作流 {route_flag}: {e}")
|
||
return {}
|
||
|
||
print("zigongggggggggggggggggggggggg")
|
||
|
||
try:
|
||
import inspect
|
||
sig = inspect.signature(workflow_func)
|
||
params = list(sig.parameters.keys())
|
||
|
||
kwargs = {
|
||
"extracted_text": extracted_text,
|
||
"combined_query": combined_query,
|
||
"history_message": history_message,
|
||
"route_flag": route_flag,
|
||
}
|
||
|
||
if "chat_id" in params:
|
||
kwargs["chat_id"] = chat_id
|
||
if "message_id" in params:
|
||
kwargs["message_id"] = message_id
|
||
if "checkpointer" in params:
|
||
kwargs["checkpointer"] = checkpointer
|
||
|
||
result = await workflow_func(**kwargs)
|
||
except Exception as e:
|
||
print(e)
|
||
logger.error(f"工作流执行失败: {e}")
|
||
return {}
|
||
|
||
return result or {}
|
||
|
||
|
||
async def stream_main_agent_execution(
|
||
raw_input: Dict[str, Any],
|
||
route_flag: str = "",
|
||
rag_prompt: str = "",
|
||
chat_id: str = "",
|
||
message_id: str = ""
|
||
) -> AsyncGenerator[str, None]:
|
||
"""
|
||
流式执行主Agent并输出函数调用信息（使用装饰器事件）
|
||
|
||
Args:
|
||
raw_input: 原始输入数据
|
||
route_flag: 路由标志
|
||
rag_prompt: RAG 提示词
|
||
chat_id: 聊天会话 ID
|
||
message_id: 消息 ID
|
||
|
||
Yields:
|
||
格式化的流式事件字符串
|
||
"""
|
||
if not chat_id:
|
||
chat_id = str(uuid.uuid4())
|
||
if not message_id:
|
||
message_id = str(uuid.uuid4())
|
||
|
||
stream_handler, event_queue = create_stream_event_handler()
|
||
register_event_callback(stream_handler)
|
||
|
||
try:
|
||
execution_done = asyncio.Event()
|
||
execution_error = None
|
||
final_result = None
|
||
checkpointer = None
|
||
|
||
async def run_agent():
|
||
nonlocal execution_error, final_result, checkpointer
|
||
route_flag_value: str = ""
|
||
try:
|
||
from checkpointer_config import checkpointer_manager
|
||
checkpointer = await checkpointer_manager.get_async_checkpointer()
|
||
|
||
main_agent = create_main_agent()
|
||
|
||
initial_route_flag: str = route_flag if (route_flag and route_flag in VALID_ROUTE_FLAGS) else ""
|
||
|
||
initial_state = {
|
||
"raw_input": raw_input,
|
||
"input_type": "",
|
||
"has_query":False,
|
||
"has_image": False,
|
||
"has_audio": False,
|
||
"has_file_text": False,
|
||
"history_message":raw_input.get("history_message", ""),
|
||
"extracted_text": "",
|
||
"image_description": "",
|
||
"asr_text": "",
|
||
"file_text": raw_input.get("file_text", ""),
|
||
"combined_query": "",
|
||
"task_type": "",
|
||
"task_classification_result": None,
|
||
"workflow_result": None,
|
||
"final_response": "",
|
||
"error_message": "",
|
||
"route_flag": initial_route_flag,
|
||
"route_params": {}
|
||
}
|
||
|
||
main_result = await main_agent.ainvoke(initial_state)
|
||
|
||
route_flag_value = (main_result.get("route_flag")or main_result.get("task_type")or "")
|
||
|
||
route_params = main_result.get("route_params") or {"combined_query": main_result.get("combined_query", "")
|
||
or main_result.get("extracted_text", ""),"extracted_text": main_result.get("extracted_text", "")}
|
||
|
||
if "file_text" in raw_input:
|
||
route_params["file_text"] = raw_input["file_text"]
|
||
|
||
if "history_message" not in route_params and "history_message" in raw_input:
|
||
route_params["history_message"] = filter_image_urls(raw_input["history_message"])
|
||
|
||
print("历史记录",route_params["history_message"])
|
||
print("用户问题", route_params["combined_query"])
|
||
print("ttttttt", route_params["extracted_text"])
|
||
|
||
sub_result = await execute_workflow_chain(
|
||
route_flag=route_flag_value,
|
||
route_params=route_params,
|
||
chat_id=chat_id,
|
||
message_id=message_id,
|
||
checkpointer=checkpointer
|
||
)
|
||
|
||
processed_content = ""
|
||
if sub_result.get("response"):
|
||
processed_content = sub_result.get("response", "")
|
||
|
||
else:
|
||
logger.warning("子agent没有返回response")
|
||
processed_content = ""
|
||
|
||
final_result = {
|
||
"content": processed_content,
|
||
"actions": sub_result.get("actions", []),
|
||
"result_tag": sub_result.get("result_tag", ""),
|
||
"suggestedReplies": sub_result.get("suggestedReplies", [])
|
||
}
|
||
except Exception as e:
|
||
execution_error = e
|
||
finally:
|
||
execution_done.set()
|
||
await event_queue.put({"type": "execution_complete"})
|
||
|
||
agent_task = asyncio.create_task(run_agent())
|
||
|
||
execution_complete_received = False
|
||
rag_result = []
|
||
graph_result = []
|
||
while True:
|
||
try:
|
||
event = await asyncio.wait_for(event_queue.get(), timeout=0.1)
|
||
|
||
if event.get("type") == "execution_complete":
|
||
execution_complete_received = True
|
||
continue
|
||
|
||
event_type = event.get("type")
|
||
title = event.get("title", "")
|
||
details = event.get("details", "")
|
||
result = event.get("result")
|
||
|
||
if event_type == "function_execution":
|
||
logger.info(f"function_execution event - title: {title}, has_result: {result is not None}, result_type: {type(result)}")
|
||
if title == "知识库搜索工具" and result and isinstance(result, dict):
|
||
print("===============================", result)
|
||
rag_result_raw = result.get("sourceCitation", {})
|
||
rag_result = {}
|
||
for doc_name, chunks in rag_result_raw.items():
|
||
if isinstance(chunks, list):
|
||
rag_result[doc_name] = [
|
||
{k: v for k, v in item.items() if k != "text"}
|
||
if isinstance(item, dict) else item
|
||
for item in chunks
|
||
]
|
||
else:
|
||
rag_result[doc_name] = chunks
|
||
elif title == "图谱检索工具" and result and isinstance(result, dict):
|
||
print("===============================", result)
|
||
graph_result = result.get("xxxx.graph", [])
|
||
else:
|
||
yield format_stream_event("function_end", {
|
||
"title": title,
|
||
"details": details
|
||
}, chat_id=chat_id, message_id=message_id)
|
||
|
||
elif event_type == "function_error":
|
||
yield format_stream_event("function_error", {
|
||
"title": title,
|
||
"error": details,
|
||
"details": details
|
||
}, chat_id=chat_id, message_id=message_id)
|
||
|
||
elif event_type == "stream_content":
|
||
content = event.get("content", "")
|
||
|
||
yield format_stream_event("stream_content", {
|
||
"content": content
|
||
}, chat_id=chat_id, message_id=message_id)
|
||
|
||
|
||
except asyncio.TimeoutError:
|
||
if execution_complete_received:
|
||
break
|
||
if execution_done.is_set() and event_queue.empty():
|
||
break
|
||
continue
|
||
|
||
await agent_task
|
||
|
||
if execution_error:
|
||
yield format_stream_event("error", {
|
||
"error": str(execution_error)
|
||
}, chat_id=chat_id, message_id=message_id)
|
||
else:
|
||
title = await extract_conversation_title(final_result.get("content", "") if final_result else "")
|
||
yield format_stream_event("execution_complete", {
|
||
"final_result": final_result
|
||
}, chat_id=chat_id, message_id=message_id)
|
||
|
||
yield format_stream_event("execution_complete", {
|
||
"rag_result": rag_result,
|
||
"graph_result": graph_result,
|
||
"title": title
|
||
}, chat_id=chat_id, message_id=message_id)
|
||
|
||
except Exception as e:
|
||
yield format_stream_event("error", {
|
||
"error": str(e)
|
||
}, chat_id=chat_id, message_id=message_id)
|
||
finally:
|
||
unregister_event_callback(stream_handler)
|
||
clear_current_stream_handler()
|
||
|
||
# =============== API 接口 ===============
|
||
@app.post("/api/agent/stream_process")
|
||
async def stream_process_request(request: AgentRequest):
|
||
"""
|
||
流式处理 Agent 请求，实时输出节点执行信息
|
||
不保存历史，调用端会自己保存历史对话信息
|
||
每次请求时创建新的主Agent实例
|
||
"""
|
||
|
||
print(f"🔍 [调试] 接收到的原始请求: {request.model_dump_json(indent=2)}")
|
||
|
||
print(f"[DEBUG] 收到请求 chat_id={request.chat_id!r}, message_id={request.message_id!r}, route_flag={request.route_flag!r}, query={str(request.query)[:100]}")
|
||
# 1. 构建当前请求的输入数据
|
||
raw_input = {}
|
||
if request.query:
|
||
if isinstance(request.query, str):
|
||
raw_input["query"] = request.query
|
||
elif isinstance(request.query, list):
|
||
for item in request.query:
|
||
if item.get("type") == "text":
|
||
raw_input["query"] = item.get("text")
|
||
print(f"当前查询: {raw_input['query']}")
|
||
if request.image:
|
||
raw_input["image"] = request.image
|
||
if request.audio:
|
||
# audio 字段支持 audio 或 asr，统一使用 audio 键
|
||
raw_input["audio"] = request.audio
|
||
# [新增] 提取 file_text
|
||
if request.file_text:
|
||
raw_input["file_text"] = request.file_text[:5000]
|
||
|
||
# 历史消息预处理：解析、移除当前用户消息、序列化
|
||
if request.history_message:
|
||
raw_input["history_message"] = preprocess_from_request(request.history_message, exclude_last=True)
|
||
|
||
# 2. 验证至少有一个输入
|
||
if not raw_input:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="至少需要提供一个输入：query、image 或 audio"
|
||
)
|
||
|
||
# 3. 获取路由标志（如果提供，将跳过任务分类步骤）
|
||
# 始终传递 route_flag，默认值为空字符串
|
||
route_flag = request.route_flag or ""
|
||
print("打印agent对应的route_flag：")
|
||
print(route_flag)
|
||
rag_prompt = process_rag_prompt(request.rag_prompt) if request.rag_prompt else ""
|
||
chat_id = request.chat_id or ""
|
||
message_id = request.message_id or ""
|
||
|
||
# 记录智能体使用情况
|
||
if route_flag and chat_id:
|
||
from tools.agent_usage_statistics import record_agent_usage
|
||
asyncio.create_task(record_agent_usage(
|
||
chat_id=chat_id,
|
||
route_flag=route_flag,
|
||
message_id=message_id
|
||
))
|
||
logger.info(f"记录智能体使用: chat_id={chat_id}, route_flag={route_flag}")
|
||
|
||
|
||
|
||
# 4. 设置当前请求的用户配置和知识库参数（线程安全，支持多用户并发）
|
||
# 如果请求中提供了这些参数，则使用请求中的值；否则使用默认值
|
||
set_request_user_config(
|
||
x_user_id=request.x_user_id,
|
||
x_user_name=request.x_user_name,
|
||
x_role=request.x_role,
|
||
text_kb_id=request.text_kb_id,
|
||
text_file_id=request.text_file_id
|
||
)
|
||
|
||
#5. 执行主Agent
|
||
async def event_generator():
|
||
async for event in stream_main_agent_execution(
|
||
raw_input=raw_input,
|
||
route_flag=route_flag,
|
||
rag_prompt=rag_prompt,
|
||
chat_id=chat_id,
|
||
message_id=message_id
|
||
):
|
||
yield event
|
||
|
||
return StreamingResponse(event_generator(), media_type="text/event-stream")
|
||
|
||
|
||
@app.get("/api/download/{file_name}")
|
||
async def download_file(file_name: str):
|
||
"""
|
||
文件下载接口
|
||
|
||
Args:
|
||
file_name: 文件名
|
||
|
||
Returns:
|
||
文件下载响应
|
||
"""
|
||
pdf_path = os.path.join(PDF_DIR, file_name)
|
||
word_path = os.path.join(WORD_DIR, file_name)
|
||
file_path = pdf_path if os.path.exists(pdf_path) else word_path
|
||
print(f"下载文件：{file_path}")
|
||
# 检查文件是否存在
|
||
ext = os.path.splitext(file_name)[1].lower()
|
||
media_type = {
|
||
".pdf": "application/pdf",
|
||
".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||
".doc": "application/msword",
|
||
}.get(ext, "application/octet-stream")
|
||
|
||
# 返回文件
|
||
return FileResponse(
|
||
path=file_path,
|
||
filename=file_name,
|
||
media_type=media_type
|
||
)
|
||
|
||
|
||
@app.get("/api/health")
|
||
async def health_check():
|
||
"""健康检查接口"""
|
||
return {
|
||
"status": "healthy",
|
||
"message": "服务正常运行，每次请求时创建新的Agent实例"
|
||
}
|
||
|
||
|
||
|
||
@app.post("/api/v1/feedback/classify")
|
||
async def feedback_classify(request: FeedbackClassifyRequest):
|
||
"""
|
||
反馈内容分类接口
|
||
|
||
功能：
|
||
1. 从反馈文本中提取设备名称
|
||
2. 根据设备名称反推系统名称
|
||
3. 对反馈内容进行分类
|
||
|
||
Args:
|
||
request: 包含 feedback_text 字段的请求体
|
||
|
||
Returns:
|
||
{
|
||
"success": true,
|
||
"system": "系统名称",
|
||
"category": "分类结果"
|
||
}
|
||
|
||
分类类别：
|
||
- 信息过时：反映文档或手册中的信息已经过时，不再适用
|
||
- 步骤错误：反映操作步骤或流程存在错误
|
||
- 步骤缺失：反映缺少必要的操作步骤或流程说明
|
||
- 备件信息有误：反映备件型号、规格、数量等信息有误
|
||
- 安全提示不足：反映缺少必要的安全警示或注意事项
|
||
- 图示不清：反映图片、示意图不清晰或难以理解
|
||
- 其他：不属于以上类别
|
||
"""
|
||
from tools.fault_statistics import analyze_feedback_and_classify
|
||
|
||
result = await analyze_feedback_and_classify(request.feedback_text)
|
||
|
||
if not result.get("success", False):
|
||
raise HTTPException(
|
||
status_code=500,
|
||
detail=result.get("error", "分类失败")
|
||
)
|
||
|
||
return result
|
||
|
||
class PdfContentItem(BaseModel):
|
||
page_idx: int
|
||
bbox: Optional[str] = None
|
||
text: Optional[str] = None
|
||
text_level: Optional[int] = None
|
||
img_path: Optional[str] = None
|
||
table_body: Optional[str] = None
|
||
|
||
|
||
class RequestWrapper(BaseModel):
|
||
pdf_contents: List[PdfContentItem]
|
||
|
||
|
||
|
||
|
||
@app.post("/split_content_list")
|
||
async def split_content_list(request_data: RequestWrapper):
|
||
pdf_contents = request_data.pdf_contents
|
||
processed_list = []
|
||
|
||
for item in pdf_contents:
|
||
# 将 Pydantic 模型转换为字典，方便修改
|
||
item_dict = item.model_dump()
|
||
|
||
# --- 核心逻辑：判断并添加 type 字段 ---
|
||
item_type = None
|
||
|
||
# 1. 如果 text 非空 -> type: "text"
|
||
if item.text:
|
||
item_type = "text"
|
||
|
||
# 2. 如果 table_body 非空 -> type: "table"
|
||
# 注意：如果 text 和 table_body 都有，根据代码顺序，table 会覆盖 text。
|
||
# 如果你的业务逻辑是互斥的或需要优先级，请调整此处顺序。
|
||
elif item.table_body:
|
||
item_type = "table"
|
||
|
||
# 3. 如果 img_path 非空 且 table_body 为空 -> type: "image"
|
||
# 注意：上面用了 elif 判断 table_body，所以到这里 table_body 肯定为空，
|
||
# 但为了逻辑清晰，显式写出条件。
|
||
elif item.img_path and not item.table_body:
|
||
item_type = "image"
|
||
|
||
# 如果匹配到了类型，写入字典
|
||
if item_type:
|
||
item_dict["type"] = item_type
|
||
|
||
processed_list.append(item_dict)
|
||
slices = get_chunk_bbox(processed_list)
|
||
slices_check = chunk_check(slices, 8000)
|
||
|
||
return {"code": 200, "message": "ok", "data": {"slices": slices_check}}
|
||
|
||
|
||
@app.post("/split_result")
|
||
async def split_result(
|
||
file: UploadFile = File(...),
|
||
image_prefix: str = Form("/api/v1/knowledge/files/images/"),
|
||
):
|
||
if not file.filename:
|
||
raise HTTPException(status_code=400, detail="File name is missing")
|
||
|
||
chunk_size = 1024
|
||
chunk_overlap = 100
|
||
|
||
filename = file.filename
|
||
suffix = PathLib(filename).suffix.lower() # 获取文件后缀
|
||
temp_input_path: Optional[PathLib] = None
|
||
converted_pdf_path: Optional[PathLib] = None
|
||
logger.info("调用知识切分接口")
|
||
|
||
try:
|
||
# ? 创建临时文件时指定后缀名
|
||
with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp_in:
|
||
content = await file.read()
|
||
logger.info(f"文件大小:{len(content)}")
|
||
logger.info(f"前10字节:{content[:10]}")
|
||
# if not content.startwith(b"%PDF"):
|
||
# logger.error("不是合法PDF")
|
||
# raise HTTPException(status_code=400,detail="Uploaded file is not a valid PDF")
|
||
tmp_in.write(content)
|
||
temp_input_path = PathLib(tmp_in.name)
|
||
# 2. 根据文件类型处理
|
||
if suffix == ".pdf":
|
||
logger.info("调用pdf切分方式")
|
||
|
||
result_data = await process_pdf_file(temp_input_path, image_prefix,filename)
|
||
if result_data is None:
|
||
raise HTTPException(status_code=502, detail="PDF processing failed")
|
||
elif suffix == ".docx":
|
||
logger.info("调用docx切分方式")
|
||
logger.info(f"转换 DOCX -> PDF: {filename}")
|
||
|
||
# 执行转换
|
||
converter.convert(str(temp_input_path), output_dir=str(DATA_DIR))
|
||
|
||
# ? 使用临时文件的基础名查找 PDF
|
||
temp_base_name = temp_input_path.stem # 如 tmp2bb0l4cn
|
||
converted_pdf_path = PathLib(DATA_DIR) / f"{temp_base_name}.pdf"
|
||
|
||
if not converted_pdf_path.exists():
|
||
raise HTTPException(status_code=500, detail=f"PDF 转换失败: {converted_pdf_path}")
|
||
|
||
result_data = await process_pdf_file(converted_pdf_path, image_prefix)
|
||
if result_data is None:
|
||
raise HTTPException(status_code=502, detail="DOCX to PDF processing failed")
|
||
|
||
elif suffix in (".txt", ".md"):
|
||
logger.info("调用txt切分方式")
|
||
|
||
# 直接读取文本内容
|
||
async with aiofiles.open(temp_input_path, "r", encoding="utf-8", errors="ignore") as f:
|
||
text_content = await f.read()
|
||
contents = split_text_preserve_sentences(text_content)
|
||
# 构造统一结构（无 images）
|
||
slices = []
|
||
for ins in contents:
|
||
slices.append({"content": ins, "positions": []})
|
||
result_data = {"slices": slices, "images": {}}
|
||
elif suffix in [".xlsx"]:
|
||
# converter.convert(str(temp_input_path), output_dir=str(DATA_DIR))
|
||
# temp_base_name = temp_input_path.stem # 如 tmp2bb0l4cn
|
||
# converted_pdf_path = PathLib(DATA_DIR) / f"{temp_base_name}.pdf"
|
||
# if not converted_pdf_path.exists():
|
||
# raise HTTPException(status_code=500, detail=f"PDF 转换失败: {converted_pdf_path}")
|
||
logger.info("调用xlsx切分方式")
|
||
|
||
result_data = await process_other_file(temp_input_path, image_prefix,filename)
|
||
if result_data is None:
|
||
raise HTTPException(status_code=502, detail="DOCX parse failed")
|
||
elif suffix in [".ppt", ".pptx"]:
|
||
logger.info("调用 PPT/PPTX 切分方式")
|
||
|
||
# 1. 判断是否为旧版 .ppt 格式，如果是，则先进行转换
|
||
if suffix == ".ppt":
|
||
logger.info(f"检测到旧版 PPT 文件，正在转换为 PPTX: {temp_input_path}")
|
||
try:
|
||
# 定义 LibreOffice 转换命令
|
||
command = [
|
||
'libreoffice',
|
||
'--headless', # 无头模式，不弹出图形界面
|
||
'--convert-to', 'pptx',
|
||
'--outdir', os.path.dirname(temp_input_path),
|
||
temp_input_path
|
||
]
|
||
|
||
# 2. 使用 run_in_executor 将阻塞的同步转换操作放入线程池，避免阻塞异步事件循环
|
||
loop = asyncio.get_event_loop()
|
||
result = await loop.run_in_executor(
|
||
None,
|
||
lambda: subprocess.run(command, capture_output=True, text=True)
|
||
)
|
||
|
||
# 3. 检查转换是否成功
|
||
if result.returncode != 0:
|
||
logger.error(f"PPT 转换失败: {result.stderr}")
|
||
raise HTTPException(status_code=500, detail="PPT to PPTX conversion failed")
|
||
|
||
# 4. 转换成功后，更新输入路径为新生成的 .pptx 文件路径
|
||
temp_input_path = os.path.splitext(temp_input_path)[0] + ".pptx"
|
||
logger.info(f"PPT 转换成功，新文件路径: {temp_input_path}")
|
||
|
||
except Exception as e:
|
||
logger.exception(f"PPT 转换过程发生异常: {e}")
|
||
raise HTTPException(status_code=500, detail=f"PPT conversion error: {str(e)}")
|
||
|
||
# 5. 无论是原本就是 .pptx，还是刚刚转换完成的 .pptx，都统一走后续处理逻辑
|
||
result_data = await process_other_file(temp_input_path, image_prefix, filename)
|
||
if result_data is None:
|
||
raise HTTPException(status_code=502, detail="PPTX parse failed")
|
||
else:
|
||
raise HTTPException(
|
||
status_code=400, detail="Unsupported file type. Only PDF, DOCX, TXT, and MD are supported."
|
||
)
|
||
|
||
return JSONResponse({"code": 200, "message": "ok", "data": result_data})
|
||
|
||
except requests.RequestException as e:
|
||
logger.error(f"Split service request failed: {e}")
|
||
raise HTTPException(status_code=502, detail=f"Split service error: {str(e)}")
|
||
except Exception as e:
|
||
logger.error(f"Internal error: {e}")
|
||
raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
|
||
finally:
|
||
# 清理临时文件
|
||
for path in [temp_input_path, converted_pdf_path]:
|
||
if path and path.exists():
|
||
try:
|
||
path.unlink()
|
||
except Exception as e:
|
||
logger.warning(f"Failed to delete temp file {path}: {e}")
|
||
|
||
|
||
@app.get("/")
|
||
async def root():
|
||
"""根路径"""
|
||
return {
|
||
"message": "Agent API 服务",
|
||
"docs": "/docs"
|
||
}
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
|
||
uvicorn.run(app, host="0.0.0.0", port=9090)
|