ai_learning/林兆新/2/sse_app.py

#!/usr/bin/env python3
"""
SSE + FastAPI 实时对话窗口
用户输入消息，后端随机生成回复
"""

import asyncio
import json
import time
import random
import os
from datetime import datetime
from typing import List, Dict, Any
from concurrent.futures import ThreadPoolExecutor
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse, HTMLResponse
import uvicorn
import dotenv

# Agent相关导入
try:
    from agno.agent import Agent
    from agno.memory.v2.db.sqlite import SqliteMemoryDb
    from agno.memory.v2.memory import Memory
    from agno.models.openai import OpenAILike
    from agno.storage.sqlite import SqliteStorage
    AGENT_AVAILABLE = True
    print("✅ Agent依赖已加载")
except ImportError as e:
    print(f"⚠️ Agent依赖未安装: {e}")
    AGENT_AVAILABLE = False

# 加载环境变量
dotenv.load_dotenv()

# 全局消息队列
message_queue: List[Dict[str, Any]] = []

# 全局Agent实例和Memory
global_agent = None
global_memory = None
global_user_id = "user_web_chat"
# 线程池执行器
thread_executor = ThreadPoolExecutor(max_workers=2)

# Agent工具函数
def get_current_time():
    """获取当前时间"""
    return f"当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

def get_user_info():
    """获取用户信息"""
    return "用户信息: 当前用户正在使用web实时对话窗口"

async def call_agent_async(agent, message, user_id):
    """异步调用Agent，避免阻塞事件循环"""
    loop = asyncio.get_event_loop()
    try:
        print(f"🔄 开始异步调用Agent... (消息: {message[:50]})")
        
        # 在线程池中执行同步的Agent调用，设置超时
        response = await asyncio.wait_for(
            loop.run_in_executor(
                thread_executor, 
                lambda: agent.run(message, user_id=user_id)
            ),
            timeout=30.0  # 30秒超时
        )
        
        print(f"✅ Agent异步调用完成")
        return response
    except asyncio.TimeoutError:
        print(f"⏰ Agent调用超时 (30秒)")
        return None
    except Exception as e:
        print(f"❌ Agent异步调用失败: {e}")
        return None

def create_agent():
    """创建具有Memory功能的Agent实例"""
    global global_agent, global_memory
    
    if not AGENT_AVAILABLE:
        print("❌ Agent依赖不可用，将使用随机回复")
        return None
    
    # 检查环境变量
    api_key = os.getenv("BAILIAN_API_KEY")
    base_url = os.getenv("BAILIAN_API_BASE_URL")
    
    if not api_key or not base_url:
        print("⚠️ 环境变量未设置，Agent功能将不可用，使用随机回复")
        return None
    
    try:
        print("🚀 创建具有Memory功能的Agent实例...")
        
        # 创建模型
        model = OpenAILike(
            id="qwen3-32b",
            api_key=api_key,
            base_url=base_url,
            request_params={"extra_body": {"enable_thinking": False}},
        )
        
        # 数据库文件
        db_file = "tmp/agent_memory.db"
        os.makedirs("tmp", exist_ok=True)
        
        # 初始化Memory v2
        memory = Memory(
            model=model,  # 使用相同的模型进行记忆管理
            db=SqliteMemoryDb(table_name="user_memories", db_file=db_file),
        )
        
        # 初始化存储
        storage = SqliteStorage(table_name="agent_sessions", db_file=db_file)
        
        # 定义记忆工具函数
        def remember_info(info: str):
            """主动记住信息的工具函数"""
            return f"我已经记住了这个信息: {info}"
        
        # 创建Agent with Memory功能
        agent = Agent(
            model=model,
            # Store memories in a database
            memory=memory,
            # Give the Agent the ability to update memories
            enable_agentic_memory=True,
            # Run the MemoryManager after each response
            enable_user_memories=True,
            # Store the chat history in the database
            storage=storage,
            # Add the chat history to the messages
            add_history_to_messages=True,
            # Number of history runs to include
            num_history_runs=3,
            # Tools
            tools=[get_current_time, get_user_info, remember_info],
            markdown=False,  # 简单文本回复
            show_tool_calls=False,  # 关闭工具调用显示，避免影响web显示
            instructions="""
你是一个具有记忆功能的友好AI助手，正在通过web实时对话窗口与用户交流。

🧠 **记忆功能**:
- 你可以记住用户的姓名、偏好和兴趣
- 保持跨会话的对话连贯性
- 基于历史对话提供个性化建议
- 记住之前讨论过的话题

💬 **对话原则**:
- 使用简洁、自然的中文回答
- 语气友好、热情
- 回答要有帮助性
- 可以调用工具获取信息
- 主动记住重要信息
- 基于记忆提供个性化回应

🎯 **个性化服务**:
- 如果用户告诉你他们的姓名，主动记住
- 记住用户的偏好和兴趣
- 在后续对话中引用之前的内容
- 提供基于历史的个性化建议

请与用户进行愉快的对话！我会记住我们的每次交流。
            """,
        )
        
        global_agent = agent
        global_memory = memory
        print("✅ Memory Agent创建成功！")
        print(f"📱 模型: qwen3-32b")
        print(f"🧠 记忆: SQLite数据库 ({db_file})")
        print(f"💾 存储: 会话历史记录")
        print(f"👤 用户ID: {global_user_id}")
        
        # 简单测试Agent是否正常工作
        try:
            test_response = agent.run("你好", user_id=global_user_id)
            print(f"🧪 Agent测试成功: {str(test_response)[:50]}...")
        except Exception as e:
            print(f"⚠️ Agent测试失败: {e}")
        
        return agent
        
    except Exception as e:
        print(f"❌ Agent创建失败: {e}")
        return None

# 随机回复内容（Agent不可用时的备用）
RANDOM_REPLIES = [
    "这是一个有趣的观点！",
    "我完全同意你的看法。",
    "让我想想这个问题...",
    "你说得很有道理。",
    "这让我想到了另一个话题。",
    "非常好的问题！",
    "我觉得你可以试试这样做。",
    "这确实是个挑战。",
    "你的想法很有创意！",
    "我需要更多信息来帮助你。",
    "这个话题很深入呢。",
    "你考虑得很周全。"
]

# 创建FastAPI应用
app = FastAPI(title="SSE实时对话", description="简单的实时聊天系统", version="1.0.0")

# 应用启动时初始化Agent
@app.on_event("startup")
async def startup_event():
    print("🚀 启动SSE实时对话系统...")
    print("📍 访问地址: http://localhost:8000")
    
    # 初始化Agent
    try:
        create_agent()
        
        if global_agent:
            print("✅ Memory Agent已就绪，将提供具有记忆功能的智能回复")
            print("🧠 记忆功能: 可记住用户信息和对话历史")
            print("💬 特殊命令: 在对话中输入 '记忆' 查看记忆状态")
        else:
            print("⚠️ Agent不可用，将使用随机回复")
            
    except Exception as e:
        print(f"❌ Agent创建过程中出错: {e}")
        print("⚠️ 系统将使用随机回复模式")

@app.get("/")
async def home():
    """主页 - 对话界面"""
    html_content = """
    <!DOCTYPE html>
    <html lang="zh-CN">
    <head>
        <meta charset="UTF-8">
        <meta name="viewport" content="width=device-width, initial-scale=1.0">
        <title>实时对话窗口</title>
        <style>
            * {
                margin: 0;
                padding: 0;
                box-sizing: border-box;
            }
            
            body {
                font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
                background: #f0f2f5;
                height: 100vh;
                display: flex;
                justify-content: center;
                align-items: center;
            }
            
            .chat-container {
                width: 600px;
                height: 500px;
                background: white;
                border-radius: 10px;
                box-shadow: 0 4px 20px rgba(0,0,0,0.1);
                display: flex;
                flex-direction: column;
                overflow: hidden;
            }
            
            .chat-header {
                background: #4a90e2;
                color: white;
                padding: 15px 20px;
                text-align: center;
                font-size: 18px;
                font-weight: 600;
            }
            
            .status-bar {
                padding: 8px 15px;
                background: #e8f4f8;
                font-size: 12px;
                color: #5a5a5a;
                border-bottom: 1px solid #e0e0e0;
            }
            
            .messages-container {
                flex: 1;
                padding: 15px;
                overflow-y: auto;
                background: #fafafa;
            }
            
            .message {
                margin-bottom: 12px;
                display: flex;
                animation: slideIn 0.3s ease-out;
            }
            
            @keyframes slideIn {
                from {
                    opacity: 0;
                    transform: translateY(10px);
                }
                to {
                    opacity: 1;
                    transform: translateY(0);
                }
            }
            
            .message.user {
                justify-content: flex-end;
            }
            
            .message-bubble {
                max-width: 75%;
                padding: 10px 15px;
                border-radius: 18px;
                word-wrap: break-word;
                position: relative;
            }
            
            .message.user .message-bubble {
                background: #4a90e2;
                color: white;
            }
            
            .message.bot .message-bubble {
                background: white;
                color: #333;
                border: 1px solid #e0e0e0;
                box-shadow: 0 1px 3px rgba(0,0,0,0.1);
            }
            
            .message.system .message-bubble {
                background: #fff3cd;
                color: #856404;
                border: 1px solid #ffeaa7;
                font-style: italic;
                text-align: center;
                max-width: 100%;
            }
            
            .message-time {
                font-size: 10px;
                color: #999;
                margin-top: 3px;
                text-align: right;
            }
            
            .message.user .message-time {
                text-align: right;
            }
            
            .message.bot .message-time {
                text-align: left;
            }
            
            .input-container {
                padding: 15px;
                background: white;
                border-top: 1px solid #e0e0e0;
                display: flex;
                gap: 10px;
            }
            
            .message-input {
                flex: 1;
                padding: 10px 15px;
                border: 1px solid #ddd;
                border-radius: 20px;
                outline: none;
                font-size: 14px;
                transition: border-color 0.3s;
            }
            
            .message-input:focus {
                border-color: #4a90e2;
                box-shadow: 0 0 0 2px rgba(74, 144, 226, 0.2);
            }
            
            .send-button {
                background: #4a90e2;
                color: white;
                border: none;
                border-radius: 20px;
                padding: 10px 20px;
                cursor: pointer;
                font-size: 14px;
                font-weight: 600;
                transition: all 0.3s;
            }
            
            .send-button:hover {
                background: #357abd;
                transform: translateY(-1px);
            }
            
            .send-button:disabled {
                background: #ccc;
                cursor: not-allowed;
                transform: none;
            }
            
            .typing-indicator {
                display: none;
                padding: 10px 15px;
                color: #666;
                font-style: italic;
                font-size: 12px;
            }
            
            .typing-dots {
                display: inline-block;
            }
            
            .typing-dots::after {
                content: '';
                animation: typing 1.5s infinite;
            }
            
            @keyframes typing {
                0%, 60%, 100% { content: ''; }
                30% { content: '.'; }
                40% { content: '..'; }
                50% { content: '...'; }
            }
        </style>
    </head>
    <body>
        <div class="chat-container">
            <div class="chat-header">
                💬 实时对话窗口
            </div>
            
            <div class="status-bar" id="statusBar">
                正在连接...
            </div>
            
            <div class="messages-container" id="messagesContainer">
                <!-- 消息显示区域 -->
            </div>
            
            <div class="typing-indicator" id="typingIndicator">
                机器人正在输入<span class="typing-dots"></span>
            </div>
            
                         <div class="input-container">
                 <input 
                     type="text" 
                     id="messageInput" 
                     class="message-input" 
                     placeholder="输入您的消息... (输入 '记忆' 查看我的记忆状态)"
                     maxlength="500"
                 >
                 <button id="sendButton" class="send-button">发送</button>
             </div>
        </div>
        
        <script>
            let eventSource = null;
            let isConnected = false;
            
            const messagesContainer = document.getElementById('messagesContainer');
            const messageInput = document.getElementById('messageInput');
            const sendButton = document.getElementById('sendButton');
            const statusBar = document.getElementById('statusBar');
            const typingIndicator = document.getElementById('typingIndicator');
            
            // 页面加载完成后初始化
            window.addEventListener('load', function() {
                console.log('页面加载完成');
                
                // 启用输入功能
                messageInput.disabled = false;
                sendButton.disabled = false;
                messageInput.focus();
                
                                 // 显示欢迎消息
                 addMessage('🎉 欢迎使用AI实时对话窗口！我是您的智能助手，具有记忆功能，可以记住我们的对话历史。请开始聊天吧~', 'system');
                
                // 连接SSE
                connectSSE();
            });
            
            // 连接SSE
            function connectSSE() {
                try {
                    console.log('连接SSE...');
                    eventSource = new EventSource('/sse/chat');
                    
                    eventSource.onopen = function() {
                        console.log('SSE连接成功');
                        isConnected = true;
                        statusBar.textContent = '✅ 已连接 - 消息将实时更新';
                        statusBar.style.background = '#d4edda';
                        statusBar.style.color = '#155724';
                    };
                    
                    eventSource.onmessage = function(event) {
                        try {
                            const data = JSON.parse(event.data);
                            handleMessage(data);
                        } catch (e) {
                            console.error('消息解析错误:', e);
                        }
                    };
                    
                    eventSource.onerror = function() {
                        console.log('SSE连接失败');
                        isConnected = false;
                        statusBar.textContent = '⚠️ 连接失败 - 使用离线模式';
                        statusBar.style.background = '#f8d7da';
                        statusBar.style.color = '#721c24';
                    };
                    
                } catch (error) {
                    console.error('SSE初始化错误:', error);
                    statusBar.textContent = '📱 离线模式';
                }
            }
            
            // 处理消息
            function handleMessage(data) {
                console.log('收到消息:', data);
                
                switch(data.type) {
                    case 'bot_message':
                        hideTypingIndicator();
                        addMessage(data.message, 'bot');
                        break;
                    case 'system_message':
                        addMessage(data.message, 'system');
                        break;
                }
            }
            
            // 添加消息到界面
            function addMessage(content, type) {
                console.log(`添加消息: [${type}] ${content}`);
                
                const messageDiv = document.createElement('div');
                messageDiv.className = `message ${type}`;
                
                const bubbleDiv = document.createElement('div');
                bubbleDiv.className = 'message-bubble';
                bubbleDiv.textContent = content;
                
                const timeDiv = document.createElement('div');
                timeDiv.className = 'message-time';
                timeDiv.textContent = new Date().toLocaleTimeString();
                
                messageDiv.appendChild(bubbleDiv);
                messageDiv.appendChild(timeDiv);
                
                messagesContainer.appendChild(messageDiv);
                messagesContainer.scrollTop = messagesContainer.scrollHeight;
            }
            
            // 显示输入指示器
            function showTypingIndicator() {
                typingIndicator.style.display = 'block';
                messagesContainer.scrollTop = messagesContainer.scrollHeight;
            }
            
                         // 隐藏输入指示器
             function hideTypingIndicator() {
                 typingIndicator.style.display = 'none';
             }
             
             // 显示记忆状态
             async function showMemoryStatus() {
                 try {
                     showTypingIndicator();
                     
                     const response = await fetch('/api/memory');
                     const result = await response.json();
                     
                     hideTypingIndicator();
                     
                     if (result.available) {
                         let memoryInfo = `🧠 **记忆状态信息**\n\n`;
                         memoryInfo += `👤 用户ID: ${result.user_id}\n`;
                         memoryInfo += `📊 记忆数量: ${result.memory_count}\n`;
                         memoryInfo += `💾 数据库: ${result.database}\n\n`;
                         
                         if (result.recent_memories && result.recent_memories.length > 0) {
                             memoryInfo += `📝 **最近的记忆:**\n`;
                             result.recent_memories.forEach((mem, index) => {
                                 memoryInfo += `${index + 1}. ${mem.content}...\n`;
                             });
                         } else {
                             memoryInfo += `📭 暂无记忆内容\n`;
                         }
                         
                         memoryInfo += `\n💭 **记忆功能说明:**\n`;
                         memoryInfo += `- 我可以记住您的姓名、偏好和兴趣\n`;
                         memoryInfo += `- 保持跨会话的对话连贯性\n`;
                         memoryInfo += `- 基于历史对话提供个性化建议\n`;
                         memoryInfo += `- 记住之前讨论过的话题`;
                         
                         addMessage(memoryInfo, 'bot');
                     } else {
                         addMessage('❌ 记忆功能不可用: ' + result.message, 'system');
                     }
                     
                 } catch (error) {
                     hideTypingIndicator();
                     addMessage('❌ 获取记忆状态失败: ' + error.message, 'system');
                 }
             }
            
                         // 发送消息
             async function sendMessage() {
                 const message = messageInput.value.trim();
                 console.log('准备发送消息:', message);
                 
                 if (!message) {
                     console.log('消息为空');
                     return;
                 }
                 
                 // 检查特殊命令
                 if (message === '记忆' || message.toLowerCase() === 'memory') {
                     console.log('处理记忆查询命令');
                     addMessage(message, 'user');
                     messageInput.value = '';
                     await showMemoryStatus();
                     return;
                 }
                 
                 // 立即显示用户消息
                 addMessage(message, 'user');
                 messageInput.value = '';
                 
                 // 禁用输入
                 messageInput.disabled = true;
                 sendButton.disabled = true;
                 
                 // 显示输入指示器
                 showTypingIndicator();
                
                try {
                    const response = await fetch('/api/chat', {
                        method: 'POST',
                        headers: {
                            'Content-Type': 'application/json',
                        },
                        body: JSON.stringify({
                            message: message
                        })
                    });
                    
                    const result = await response.json();
                    console.log('发送结果:', result);
                    
                    if (!result.success) {
                        hideTypingIndicator();
                        addMessage('❌ 发送失败: ' + result.error, 'system');
                    }
                    
                } catch (error) {
                    console.error('发送错误:', error);
                    hideTypingIndicator();
                    addMessage('❌ 网络错误: ' + error.message, 'system');
                } finally {
                    // 重新启用输入
                    messageInput.disabled = false;
                    sendButton.disabled = false;
                    messageInput.focus();
                }
            }
            
            // 事件监听器
            sendButton.addEventListener('click', sendMessage);
            
            messageInput.addEventListener('keypress', function(e) {
                if (e.key === 'Enter' && !e.shiftKey) {
                    e.preventDefault();
                    sendMessage();
                }
            });
            
            // 页面关闭时断开连接
            window.addEventListener('beforeunload', function() {
                if (eventSource) {
                    eventSource.close();
                }
            });
        </script>
    </body>
    </html>
    """
    return HTMLResponse(content=html_content)

# SSE消息流生成器
async def generate_chat_stream():
    """生成聊天消息流"""
    client_id = f"client_{int(time.time())}"
    last_message_count = 0
    
    try:
        print(f"新的SSE连接: {client_id}")
        
        while True:
            # 检查新消息
            if len(message_queue) > last_message_count:
                for i in range(last_message_count, len(message_queue)):
                    message = message_queue[i]
                    yield f"data: {json.dumps(message, ensure_ascii=False)}\n\n"
                
                last_message_count = len(message_queue)
            
            await asyncio.sleep(0.1)  # 100ms检查间隔
            
    except Exception as e:
        print(f"SSE流错误: {e}")
    finally:
        print(f"SSE连接断开: {client_id}")

@app.get("/sse/chat")
async def sse_chat():
    """SSE聊天端点"""
    return StreamingResponse(
        generate_chat_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "Access-Control-Allow-Origin": "*",
        }
    )

@app.post("/api/chat")
async def chat_api(request: Request):
    """处理聊天消息"""
    global global_agent, global_user_id
    
    try:
        data = await request.json()
        user_message = data.get("message", "").strip()
        
        if not user_message:
            return {"success": False, "error": "消息不能为空"}
        
        print(f"📨 收到用户消息: {user_message}")
        print(f"🤖 Agent可用性: {global_agent is not None}")
        
        # 模拟思考时间
        await asyncio.sleep(random.uniform(0.5, 1.5))
        
        bot_reply = None
        
        # 尝试使用Agent生成回复
        if global_agent:
            try:
                print("🤖 调用Memory Agent生成回复...")
                
                # 异步调用Agent处理消息，传入user_id以关联记忆
                response = await call_agent_async(global_agent, user_message, global_user_id)
                
                if response:
                    bot_reply = response.content if hasattr(response, 'content') else str(response)
                    print(f"✅ Agent回复: {bot_reply}")
                    print(f"🧠 记忆已更新 (用户ID: {global_user_id})")
                else:
                    print("❌ Agent返回空响应")
                    bot_reply = None
                
            except Exception as e:
                print(f"❌ Agent调用失败: {e}")
                bot_reply = None
        
        # 如果Agent不可用或失败，使用随机回复
        if not bot_reply:
            bot_reply = random.choice(RANDOM_REPLIES)
            print(f"🎲 使用随机回复: {bot_reply}")
        
        # 确保回复不为空
        if not bot_reply:
            bot_reply = "抱歉，我暂时无法回复。请稍后再试。"
            print(f"⚠️ 使用默认回复: {bot_reply}")
        
        # 添加机器人回复到消息队列
        bot_message = {
            "type": "bot_message",
            "message": bot_reply,
            "timestamp": datetime.now().isoformat()
        }
        message_queue.append(bot_message)
        
        print(f"📤 机器人回复已添加到消息队列: {bot_reply[:50]}...")
        
        return {
            "success": True,
            "message": "消息已处理",
            "timestamp": datetime.now().isoformat()
        }
        
    except Exception as e:
        print(f"处理消息错误: {e}")
        return {"success": False, "error": f"处理错误: {str(e)}"}

@app.get("/api/status")
async def get_status():
    """获取系统状态"""
    return {
        "status": "running",
        "messages": len(message_queue),
        "timestamp": datetime.now().isoformat()
    }

@app.post("/api/clear")
async def clear_messages():
    """清空消息历史"""
    global message_queue
    message_queue.clear()
    
    # 添加清空提示消息
    clear_msg = {
        "type": "system_message",
        "message": "💬 消息历史已清空",
        "timestamp": datetime.now().isoformat()
    }
    message_queue.append(clear_msg)
    
    return {"success": True, "message": "消息历史已清空"}

@app.get("/api/memory")
async def get_memory_status():
    """获取Agent记忆状态"""
    global global_memory, global_user_id, global_agent
    
    if not global_agent or not global_memory:
        return {
            "available": False,
            "message": "Memory功能不可用",
            "user_id": global_user_id
        }
    
    try:
        # 尝试获取记忆信息
        memories = global_memory.get_user_memories(user_id=global_user_id)
        memory_count = len(memories) if memories else 0
        
        # 获取最近的3条记忆摘要
        recent_memories = []
        if memories:
            for mem in memories[-3:]:
                try:
                    # UserMemory对象的属性访问
                    content = getattr(mem, 'content', '')[:100] if hasattr(mem, 'content') else str(mem)[:100]
                    timestamp = getattr(mem, 'created_at', '') if hasattr(mem, 'created_at') else ''
                    recent_memories.append({
                        "content": content,
                        "timestamp": str(timestamp),
                    })
                except Exception as e:
                    # 备用方案：直接转换为字符串
                    recent_memories.append({
                        "content": str(mem)[:100],
                        "timestamp": "",
                    })
        
        return {
            "available": True,
            "user_id": global_user_id,
            "memory_count": memory_count,
            "recent_memories": recent_memories,
            "database": "tmp/agent_memory.db",
            "timestamp": datetime.now().isoformat()
        }
        
    except Exception as e:
        return {
            "available": True,
            "user_id": global_user_id,
            "memory_count": "unknown",
            "error": str(e),
            "message": "记忆系统已启用但获取详情失败",
            "timestamp": datetime.now().isoformat()
        }

if __name__ == "__main__":
    print("🚀 启动SSE实时对话系统...")
    print("🌐 访问地址: http://localhost:8081")
    print("🤖 支持Agent智能回复 + SSE实时推送")
    print("📋 如需Agent功能，请配置环境变量：")
    print("   BAILIAN_API_KEY=your_api_key")
    print("   BAILIAN_API_BASE_URL=your_base_url")
    print("=" * 50)
    
    uvicorn.run(
        "sse_app:app",
        host="0.0.0.0",
        port=8081,
        reload=True,
        log_level="info"
    )