#!/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 = """
实时对话窗口
"""
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"
)