"""app.py FastAPI 主 Agent 接口 - 修复重复执行问题 提供主脑 Agent 的 HTTP API 服务，支持流式输出 集成 LangGraph Checkpointer 实现状态持久化 """ import logging from time import sleep 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 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 utils.source_citations import merge_rag_source_citations from checkpointer_config import ( CheckpointerManager, checkpointer_manager ) from admin_routes import router as admin_router from wiki_engine.routes import router as wiki_router # 创建 FastAPI 应用 app = FastAPI(max_request_size=1024 * 1024 * 10) 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.include_router(wiki_router) app.include_router(admin_router) # app.mount("/picture", StaticFiles(directory="/app/picture"), name="pictures") @app.on_event("startup") async def startup_event(): """应用启动时初始化""" await checkpointer_manager.setup() # 初始化智能体使用统计表 from tools.agent_usage_statistics import init_agent_usage_table await init_agent_usage_table() from wiki_engine.db import init_wiki_tables await init_wiki_tables() from wiki_engine.worker import start_wiki_queue_workers await start_wiki_queue_workers() from tools.agent_registration import register_wiki_admin_agent await register_wiki_admin_agent() 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 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", {}) merge_rag_source_citations(rag_result, rag_result_raw) 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: 文件下载响应 """ safe_name = os.path.basename(file_name) if safe_name != file_name or safe_name in ("", ".", ".."): raise HTTPException(status_code=400, detail="Invalid file name") pdf_path = os.path.abspath(os.path.join(PDF_DIR, safe_name)) word_path = os.path.abspath(os.path.join(WORD_DIR, safe_name)) allowed_dirs = (os.path.abspath(PDF_DIR), os.path.abspath(WORD_DIR)) candidates = [] for candidate in (pdf_path, word_path): if any(os.path.commonpath([candidate, allowed_dir]) == allowed_dir for allowed_dir in allowed_dirs): candidates.append(candidate) file_path = next((path for path in candidates if os.path.isfile(path)), None) if not file_path: raise HTTPException(status_code=404, detail="File not found") print(f"下载文件：{file_path}") # 检查文件是否存在 ext = os.path.splitext(file_name)[1].lower() ext = os.path.splitext(safe_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=safe_name, media_type=media_type, headers={"X-Content-Type-Options": "nosniff"} ) @app.get("/api/health") async def health_check(): """健康检查接口""" return { "status": "healthy", "message": "服务正常运行，每次请求时创建新的Agent实例" } @app.get("/api/v1/statistics/fault-type-statistics") async def fault_type_statistics(ship_number: Optional[str] = None): """ 故障类型统计接口 统计生成报告里面的故障类型，按舷号分类并计算百分比 Args: ship_number: 舷号过滤（可选，如果提供则只统计该舷号的报告） Returns: 统计结果，包含 faultTypeData 字段 """ print("故障类型统计街路口") from tools.fault_statistics import statistics_fault_types result = await statistics_fault_types( ship_number_filter=ship_number ) if not result.get("success", False): raise HTTPException( status_code=500, detail=result.get("error", "统计失败") ) return result @app.get("/api/v1/statistics/fault-frequency-statistics") async def fault_frequency_statistics( top_n: Optional[int] = 6, time_range: Optional[str] = "全部" ): """ 故障发生次数统计接口 统计生成报告里面的故障发生次数，按频率排序 Args: top_n: 返回前 N 个高频故障（默认为 10） time_range: 时间范围（可选，默认为"全部"） - "全部": 统计所有报告 - "近七天": 统计最近7天的报告 - "近一个月": 统计最近30天的报告 - "近一年": 统计最近365天的报告 Returns: 统计结果，包含 highFreqFaults 字段 """ print("故障发生次数接口") from datetime import datetime, timedelta start_date = None end_date = None if time_range == "近七天": end_date = datetime.now().strftime("%Y-%m-%d") start_date = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") elif time_range == "近一个月": end_date = datetime.now().strftime("%Y-%m-%d") start_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d") elif time_range == "近一年": end_date = datetime.now().strftime("%Y-%m-%d") start_date = (datetime.now() - timedelta(days=365)).strftime("%Y-%m-%d") from tools.fault_statistics import statistics_fault_frequency result = await statistics_fault_frequency( top_n=top_n, start_date=start_date, end_date=end_date ) if not result.get("success", False): raise HTTPException( status_code=500, detail=result.get("error", "统计失败") ) return result @app.get("/api/v1/statistics/spare-parts-statistics") async def spare_parts_statistics( ship_number: Optional[str] = None, time_range: Optional[str] = "全部", top_n: Optional[int] = 6 ): """ 备件消耗统计接口 统计生成报告里面的备件消耗情况 Args: ship_number: 舷号过滤（可选） time_range: 时间范围（可选，默认为"全部"） - "全部": 统计所有报告 - "近七天": 统计最近7天的报告 - "近一个月": 统计最近30天的报告 - "近一年": 统计最近365天的报告 top_n: 返回前 N 个备件（默认为 5） Returns: 统计结果，包含 sparePartData 字段 """ print("备件消耗统计接口") from datetime import datetime, timedelta start_date = None end_date = None if time_range == "近七天": end_date = datetime.now().strftime("%Y-%m-%d") start_date = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") elif time_range == "近一个月": end_date = datetime.now().strftime("%Y-%m-%d") start_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d") elif time_range == "近一年": end_date = datetime.now().strftime("%Y-%m-%d") start_date = (datetime.now() - timedelta(days=365)).strftime("%Y-%m-%d") from tools.fault_statistics import statistics_spare_parts result = await statistics_spare_parts( ship_number_filter=ship_number, start_date=start_date, end_date=end_date, top_n=top_n ) if not result.get("success", False): raise HTTPException( status_code=500, detail=result.get("error", "统计失败") ) return result @app.get("/api/v1/statistics/ship-numbers") async def get_all_ship_numbers(): """ 获取所有舷号接口 从知识库树形结构中提取所有可用的舷号列表 Returns: { "success": true, "data": { "ship_numbers": ["101", "111", "122", ...], "total": 10 } } """ import httpx import re from config import KB_TREE_CONFIG KB_TREE_URL = KB_TREE_CONFIG['url'] KB_TREE_HEADERS = { "accept": "application/json", "x-user-id": "1", "x-user-name": "testuser", "x-role": "admin", } try: async with httpx.AsyncClient(timeout=30.0, verify=False) as client: params = {"kb_id": 2} response = await client.get(KB_TREE_URL, headers=KB_TREE_HEADERS, params=params) if response.status_code != 200: raise HTTPException( status_code=response.status_code, detail=f"知识库API请求失败: {response.status_code}" ) result = response.json() if isinstance(result, dict): if "code" in result and result.get("code") != 200: raise HTTPException( status_code=500, detail=f"知识库API返回错误: {result}" ) kb_data = result.get("data") or result.get("list") or {} elif isinstance(result, list): kb_data = result[0] if result else {} else: kb_data = {} children = kb_data.get("children", []) if not children: return { "success": True, "data": { "ship_numbers": [], "total": 0 } } ship_numbers_set = set() for kb in children: kb_name = kb.get("name") or "" kb_numbers = re.findall(r"\d+", kb_name) for num in kb_numbers: ship_numbers_set.add(num) ship_numbers = sorted(list(ship_numbers_set), key=lambda x: int(x)) return { "success": True, "data": { "ship_numbers": ship_numbers, "total": len(ship_numbers) } } except HTTPException: raise except Exception as e: logger.error(f"获取舷号列表失败: {str(e)}") raise HTTPException( status_code=500, detail=f"获取舷号列表失败: {str(e)}" ) @app.get("/api/v1/statistics/agent-usage-statistics") async def agent_usage_statistics(): """ 智能体使用统计接口 统计智能体的调用次数，按类型分类 Returns: { "success": true, "data": { "total": "8,794", "types": [ { "name": "操作使用", "count": "2,966", "icon": "repair", "color": "#00C2FF" }, { "name": "故障排查与修理", "count": "3,620", "icon": "detection", "color": "#FFB800" }, { "name": "舰船百科", "count": "2,208", "icon": "doc", "color": "#9153FF" } ], "ratio": [ { "name": "操作使用", "value": 34, "itemStyle": { "color": "#00C2FF" } }, { "name": "故障排查与修理", "value": 41, "itemStyle": { "color": "#FFB800" } }, { "name": "舰船百科", "value": 25, "itemStyle": { "color": "#9153FF" } } ] } } """ print("智能体调用统计接口") from tools.agent_usage_statistics import get_agent_usage_statistics result = await get_agent_usage_statistics() if not result.get("success", False): raise HTTPException( status_code=500, detail="统计失败" ) return result @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 @app.get("/") async def root(): """根路径""" return { "message": "Agent API 服务", "docs": "/docs" } # =============== 主函数 =============== if __name__ == "__main__": uvicorn.run( "app:app", host="0.0.0.0", port=9088 )