from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import uvicorn from fastapi import WebSocket import json from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware import asyncio import json import time from .Agents import XiaoLang from fastapi.staticfiles import StaticFiles import os from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from pydantic import BaseModel,Field model = ChatOpenAI( model="deepseek-ai/DeepSeek-V3", api_key="sk-pedbhsizuejmthmfylkvuppnkcpfpxumlieydxxxlxsksmmg", base_url="https://api.siliconflow.cn/v1" ) app = FastAPI() # Define the path to the static directory # This assumes the static directory is at the same level as Server.py static_directory = os.path.join(os.path.dirname(__file__), "static") # If the directory doesn't exist, create it if not os.path.exists(static_directory): os.makedirs(static_directory) # Mount the static directory app.mount("/static", StaticFiles(directory=static_directory), name="static") origins = [ "http://localhost:3000", # 允许来自本地 3000 端口的请求 "http://localhost", "http://localhost:8000", ] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 存储连接的客户端 connected_clients = set() # 存储每个客户端的上次活动时间 client_last_activity = {} # 心跳检查间隔（秒） HEARTBEAT_INTERVAL = 30 @app.post("/chat") def chat(query: str): res = XiaoLang.invoke({ "messages": [ { "role": "user", "content": query } ] },{"configurable": {"thread_id": "anonymous"}}) return res @app.websocket("/ws/chat") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() connected_clients.add(websocket) client_last_activity[websocket] = time.time() print("Connection established") is_closed = False # 创建心跳检查任务 heartbeat_task = asyncio.create_task(check_heartbeat(websocket)) try: while True: data = await websocket.receive_text() print(f"Received text: {data}") # 更新客户端活动时间 client_last_activity[websocket] = time.time() # 解析JSON数据 try: message = json.loads(data) # 检查消息类型 if message.get("type") == "ping": # 如果是心跳消息，则发送pong响应 print("Received ping, sending pong") await websocket.send_text(json.dumps({"type": "pong"})) else: # 处理其他消息 await process_message(websocket, message) except json.JSONDecodeError: print(f"Invalid JSON format received: {data}") await websocket.send_text("Invalid JSON format.") except WebSocketDisconnect: print("Connection closed") cleanup_connection(websocket) heartbeat_task.cancel() # 取消心跳任务 if not is_closed: is_closed = True try: await websocket.close() except: pass # 连接可能已经关闭，忽略异常 except Exception as e: print(f"WebSocket error: {e}") cleanup_connection(websocket) heartbeat_task.cancel() # 取消心跳任务 if not is_closed: is_closed = True try: await websocket.close() except: pass # 连接可能已经关闭，忽略异常 # 清理连接资源 def cleanup_connection(websocket: WebSocket): connected_clients.discard(websocket) client_last_activity.pop(websocket, None) # 心跳检查任务 async def check_heartbeat(websocket: WebSocket): try: while True: await asyncio.sleep(HEARTBEAT_INTERVAL) # 每30秒检查一次 current_time = time.time() # 如果客户端长时间未活动，关闭连接 if websocket in client_last_activity: last_activity = client_last_activity[websocket] if current_time - last_activity > HEARTBEAT_INTERVAL * 2: print(f"Client inactive for {current_time - last_activity} seconds, closing connection") try: await websocket.close(code=1000, reason="Inactivity timeout") except: pass # 连接可能已经关闭，忽略异常 cleanup_connection(websocket) break except asyncio.CancelledError: # 任务被取消 print("Heartbeat check task cancelled") except Exception as e: print(f"Error in heartbeat check: {e}") class Feelings(BaseModel): feeling: str = Field(..., title="情绪分类", description="用户输入的情绪分类") action : str = Field(..., title="动作", description="根据用户输入以及判断的情绪，挑选一个合适的动作，相同名字的可以随机选一个，返回给用户。") async def process_message(websocket: WebSocket, message: dict): try: # 提取消息内容和用户ID（如果有） user_message = message.get("message", "") user_id = message.get("user_id", "anonymous") voice = message.get("voice", "zh-CN-XiaomoNeural") print(f"Processing message from user {user_id}: '{user_message}'") # 边界情况处理：如果消息为空 if not user_message: # 发送一个简单的回应 response_data = { "type": "stream", "content": "我没有收到您的消息内容，请重新输入。", "voice": voice } await websocket.send_text(json.dumps(response_data)) # 发送完成信号 complete_data = { "type": "complete", "content": "我没有收到您的消息内容，请重新输入。", "voice": voice, "qingxu": "chat" } await websocket.send_text(json.dumps(complete_data)) print("Empty message handling complete") return # 接入Agents.py中的app finallMessage = "" for message_chunk,metas in XiaoLang.stream( { "messages": [ { "role": "user", "content": user_message } ] }, stream_mode="messages", config={"thread_id": user_id} ): # 检查websocket是否关闭 if websocket.client_state.name != "CONNECTED": print("WebSocket disconnected during streaming") return if message_chunk.content: finallMessage += message_chunk.content # 发送流式内容 stream_data = { "type": "stream", "content": message_chunk.content, "voice": voice } await websocket.send_text(json.dumps(stream_data)) # 等待一小段时间再发送完成信号，确保前端已经处理了所有流式内容 await asyncio.sleep(0.2) prompt = ChatPromptTemplate.from_template(""" 分析用户输入，参考下面的列表，判断用户的情绪分类，返回一个对应单词： 情绪类型对照： - default: 中性、平静的情绪状态 - upbeat: 积极向上、充满活力的情绪 - angry: 愤怒、生气的情绪 - cheerful: 开心愉快、充满欢乐的情绪 - depressed: 沮丧、压抑的情绪 - friendly: 友好、亲切的情绪 情绪分类指南： 1. default: 用于表达中性或普通的情绪状态 2. upbeat: 用于表达积极向上、充满干劲的状态 3. angry: 用于表达愤怒、不满、生气的情绪 4. cheerful: 用于表达欢快、喜悦的情绪 5. depressed: 用于表达消极、低落、压抑的情绪 6. friendly: 用于表达友善、亲切的情绪 根据用户输入以及判断的情绪，挑选一个合适的动作，相同名字的可以随机选一个，返回给用户。 第一:numeric1-left-1 第二:numeric2-left-1 第三:numeric3-left-1 笨蛋:thumbsup-left-1 向前看:show-front-1 直视:show-front-2 直视:show-front-3 直视:show-front-4 直视:show-front-5 若有所思:think-twice-1 直视:show-front-6 直视:show-front-7 直视:show-front-8 直视:show-front-9 记住：值可以返回返回动作的英文名字，例如：numeric1-left-1，不可以有其他的内容。 用户输入: {topic} {format_instructions} """) parser = PydanticOutputParser(pydantic_object=Feelings) emotion_chain = prompt | model | parser emotion_result = emotion_chain.invoke({"topic": user_message, "format_instructions": parser.get_format_instructions()}) print("用户情绪为：", emotion_result.feeling) print("用户动作为：", emotion_result.action) # 最后一定要发送完成信号 complete_data = { "type": "complete", "content": finallMessage, "voice": voice, "qingxu": emotion_result.feeling, "action": emotion_result.action } await websocket.send_text(json.dumps(complete_data)) print(f"Stream complete, sent completion signal for message: '{user_message}'") except Exception as e: print(f"Error in process_message: {e}") # 尝试发送错误响应 try: error_data = { "type": "complete", "content": f"处理您的消息时出现错误，请重试。", "voice": message.get("voice", "zh-CN-XiaomoNeural"), "qingxu": "serious" } await websocket.send_text(json.dumps(error_data)) except: print("Failed to send error response") def main(): """Entry point for the application.""" uvicorn.run(app, host="localhost", port=8000) if __name__ == "__main__": main()