import json import uuid from fastapi import Depends, APIRouter, Query, HTTPException from fastapi.responses import JSONResponse from pydantic import BaseModel from sqlalchemy.orm import Session from app.api import Response, get_current_user, ResponseList, process_files from app.api.user import reset_user_pwd from app.config.config import settings from app.models.agent_model import AgentType, AgentModel from app.models.base_model import get_db from app.models.session_model import SessionModel from app.models.user_model import UserModel from app.service.bisheng import BishengService from app.service.dialog import get_session_history from app.service.ragflow import RagflowService from app.service.service_token import get_ragflow_token, get_bisheng_token router = APIRouter() @router.get("/list", response_model=ResponseList) async def agent_list(db: Session = Depends(get_db)): agents = db.query(AgentModel).order_by(AgentModel.sort.asc()).all() result = [item.to_dict() for item in agents] return ResponseList(code=200, msg="", data=result) @router.get("/{agent_id}/sessions", response_model=ResponseList) async def chat_list( agent_id: str, page: int = Query(1, ge=1), limit: int = Query(1000, ge=1, le=1000), db: Session = Depends(get_db), current_user: UserModel = Depends(get_current_user)): agent = db.query(AgentModel).filter(AgentModel.id == agent_id).first() if not agent: return ResponseList(code=404, msg="Agent not found") if agent.agent_type == AgentType.RAGFLOW: ragflow_service = RagflowService(base_url=settings.fwr_base_url) try: token = await get_ragflow_token(db, current_user.id) result = await ragflow_service.get_chat_sessions(token, agent_id) if not result: result = await get_session_history(db, current_user.id, agent_id) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) return ResponseList(code=200, msg="", data=result) elif agent.agent_type == AgentType.BISHENG: bisheng_service = BishengService(base_url=settings.sgb_base_url) try: token = await get_bisheng_token(db, current_user.id) result = await bisheng_service.get_chat_sessions(token, agent_id, page, limit) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) return ResponseList(code=200, msg="", data=result) elif agent.agent_type == AgentType.BASIC: offset = (page - 1) * limit records = db.query(SessionModel).filter(SessionModel.agent_id == agent_id, SessionModel.tenant_id==current_user.id).order_by(SessionModel.create_date.desc()).offset(offset).limit(limit).all() result = [item.to_dict() for item in records] return ResponseList(code=200, msg="", data=result) elif agent.agent_type == AgentType.DIFY: offset = (page - 1) * limit records = db.query(SessionModel).filter(SessionModel.agent_id == agent_id, SessionModel.tenant_id==current_user.id).order_by(SessionModel.create_date.desc()).offset(offset).limit(limit).all() result = [item.to_dict() for item in records] return ResponseList(code=200, msg="", data=result) else: return ResponseList(code=200, msg="Unsupported agent type") @router.get("/{agent_id}/{conversation_id}/session_log") async def session_log(agent_id: str, conversation_id: str, db: Session = Depends(get_db), current_user: UserModel = Depends(get_current_user)): agent = db.query(AgentModel).filter(AgentModel.id == agent_id).first() if not agent: return Response(code=404, msg="Agent not found") if agent.agent_type == AgentType.RAGFLOW: ragflow_service = RagflowService(base_url=settings.fwr_base_url) try: token = await get_ragflow_token(db, current_user.id) result = await ragflow_service.get_session_log(token, conversation_id) if 'session_log' in result and 'reference' in result: combined_logs = [] last_question = None references = result['reference'] reference_index = 0 for session in result['session_log']: if session['role'] == 'user': last_question = session['message'] elif session['role'] == 'assistant' and last_question: if reference_index < len(references): reference = references[reference_index] else: reference = None combined_logs.append({ 'question': last_question, 'answer': session['message'], 'reference': reference }) last_question = None reference_index += 1 return JSONResponse(status_code=200, content={"code": 200, "data": combined_logs}) else: return JSONResponse(status_code=200, content={"code": 400, "message": "Invalid result structure"}) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) elif agent.agent_type == AgentType.BISHENG: is_join = False if agent.name == "报告生成": is_join = True bisheng_service = BishengService(base_url=settings.sgb_base_url) try: token = await get_bisheng_token(db, current_user.id) result = await bisheng_service.get_session_log(token, agent_id, conversation_id) combined_logs = [] last_question = None answer_str = "" files = [] for session in result: # print(session) # 检查 session 是否为 None if session is None: continue message = session.get('message') # 判断 message 是否是字符串，然后尝试解析为 JSON 对象 if isinstance(message, str): try: message_json = json.loads(message) if 'question' in message_json: message = message_json['question'] elif 'query' in message_json: message = message_json['query'] elif 'report_name' in message_json: message = message_json['report_name'] except Exception as e: pass # 非 JSON 字符串，继续使用原始 message if session.get('files') and isinstance(session.get('files'), str): try: files = json.loads(session.get('files')) process_files(files, agent_id) except Exception as e: pass # 非 JSON 字符串，继续使用原始 message # 检查 message 是否为 None if message is None: continue if is_join: ... if session.get('role') == 'question': last_question = message elif session.get('role') == 'answer': answer_str += message else: if session.get('role') == 'question': last_question = message elif session.get('role') == 'answer' and last_question: combined_logs.append({ 'question': last_question, 'answer': message }) last_question = None return JSONResponse(status_code=200, content={"code": 200, "data": combined_logs if combined_logs else [{'question': last_question, 'answer': answer_str, 'files': files}]}) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) elif agent.agent_type == AgentType.BASIC: data = [] session = db.query(SessionModel).filter(SessionModel.id == conversation_id).first() if session: tmp_data = {} for i in session.log_to_json().get("message", []): if i.get("role") == "user": tmp_data["question"]=i.get("content") elif i.get("role") == "assistant": if isinstance(i.get("content"), dict): tmp_data["answer"] = i.get("content", {}).get("message") if "file_name" in i.get("content", {}): tmp_data["files"] = [{"file_name":i.get("content", {}).get("file_name"), "file_url":i.get("content", {}).get("file_url")}] else: tmp_data["answer"] = i.get("content") if "excel_url" in i: tmp_data["excel_url"] = i.get("excel_url") if "image_url" in i: tmp_data["image_url"] = i.get("image_url") if "sql" in i: tmp_data["sql"] = i.get("sql") if "code" in i: tmp_data["code"] = i.get("code") if "e" in i: tmp_data["e"] = i.get("e") if "image_name" in i: tmp_data["image_name"] = i.get("image_name") if "excel_name" in i: tmp_data["excel_name"] = i.get("excel_name") data.append(tmp_data) tmp_data = {} if tmp_data: data.append(tmp_data) return JSONResponse(status_code=200, content={"code": 200, "data": data}) elif agent.agent_type == AgentType.DIFY: data = [] session = db.query(SessionModel).filter(SessionModel.id == conversation_id).first() if session: tmp_data = {} for i in session.log_to_json().get("message", []): if i.get("role") == "user": tmp_data["question"] = i.get("content") elif i.get("role") == "assistant": if isinstance(i.get("content"), dict): tmp_data["answer"] = i.get("content", {}).get("answer") if "file_name" in i.get("content", {}): tmp_data["files"] = [{"file_name": i.get("content", {}).get("file_name"), "file_url": i.get("content", {}).get("file_url")}] if "images" in i.get("content", {}): tmp_data["images"] = i.get("content", {}).get("images") if "download_url" in i.get("content", {}): tmp_data["download_url"] = i.get("content", {}).get("download_url") else: tmp_data["answer"] = i.get("content") data.append(tmp_data) tmp_data = {} if tmp_data: data.append(tmp_data) return JSONResponse(status_code=200, content={"code": 200, "data": data}) else: return JSONResponse(status_code=200, content={"code": 200, "log": "Unsupported agent type"}) @router.get("/get-chat-id/{agent_id}", response_model=Response) async def get_chat_id(agent_id: str, db: Session = Depends(get_db)): agent = db.query(AgentModel).filter(AgentModel.id == agent_id).first() if not agent: return Response(code=404, msg="Agent not found") return Response(code=200, msg="", data={"chat_id": uuid.uuid4().hex})