1025 lines
34 KiB
Python
1025 lines
34 KiB
Python
"""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
|
||
)
|
||
|
||
|
||
|
||
|
||
|