"""
Web 版 Agent 界面 - 完整显示所有信息
Flask + SocketIO 实现实时双向通信
"""
import asyncio
import json
import sys
import threading
import queue
from pathlib import Path
from datetime import datetime
from flask import Flask, render_template_string
from flask_socketio import SocketIO, emit
# 导入 DeepSeek Agent 的核心类
from yuanbao_agent_chat import SoloAgent, ExecutionEngine
app = Flask(__name__)
app.config['SECRET_KEY'] = 'deepseek-agent-secret'
socketio = SocketIO(app, cors_allowed_origins="*", async_mode='threading')
# 全局状态
agent_instance = None
browser_running = False
message_queue = queue.Queue()
# HTML 模板
HTML_TEMPLATE = '''
DeepSeek Agent - Web 控制台
'''
class WebInterface:
"""Web 界面适配器"""
def __init__(self):
self.socket = None
def set_socket(self, socket):
self.socket = socket
def emit(self, event, data):
socketio.emit(event, data)
def print_user(self, message: str):
"""显示用户消息"""
self.emit('user_message', {
'content': message,
'timestamp': datetime.now().strftime('%H:%M:%S')
})
def print_ai(self, message: str, raw: str = ""):
"""显示 AI 消息"""
self.emit('dp_message', {
'content': message,
'raw': raw,
'timestamp': datetime.now().strftime('%H:%M:%S')
})
def print_system(self, message: str):
"""显示系统消息"""
self.emit('system_message', {
'content': message,
'timestamp': datetime.now().strftime('%H:%M:%S')
})
def print_error(self, message: str):
"""显示错误消息"""
self.emit('error_message', {
'content': message,
'timestamp': datetime.now().strftime('%H:%M:%S')
})
def log(self, category: str, message: str, level: str = "info"):
"""添加日志条目"""
self.emit('log_entry', {
'time': datetime.now().strftime('%H:%M:%S'),
'category': category,
'message': message,
'level': level
})
def update_browser_status(self, running: bool, message: str):
"""更新浏览器状态"""
self.emit('browser_status', {
'running': running,
'message': message
})
# 创建 Web 界面实例
web_interface = WebInterface()
@app.route('/')
def index():
return render_template_string(HTML_TEMPLATE)
@socketio.on('connect')
def handle_connect():
web_interface.set_socket(request.sid if hasattr(request, 'sid') else None)
web_interface.update_browser_status(browser_running, '浏览器未启动')
@socketio.on('disconnect')
def handle_disconnect():
pass
@socketio.on('start_browser')
def handle_start_browser():
"""启动浏览器"""
global agent_instance, browser_running
def start():
global agent_instance, browser_running
try:
web_interface.update_browser_status(False, '正在启动浏览器...')
web_interface.log('BROWSER', '正在初始化 Playwright...')
if agent_instance is None:
agent_instance = SoloAgent()
# 创建新的事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 启动浏览器
loop.run_until_complete(agent_instance.start())
browser_running = True
web_interface.update_browser_status(True, '浏览器已连接')
web_interface.log('BROWSER', '浏览器启动成功', level='success')
web_interface.print_system('🚀 浏览器已启动，可以开始对话了')
loop.close()
except Exception as e:
browser_running = False
web_interface.update_browser_status(False, '启动失败')
web_interface.print_error(f'浏览器启动失败: {str(e)}')
web_interface.log('ERROR', str(e), level='error')
import traceback
web_interface.log('ERROR', traceback.format_exc(), level='error')
threading.Thread(target=start, daemon=True).start()
@socketio.on('stop_browser')
def handle_stop_browser():
"""停止浏览器"""
global agent_instance, browser_running
def stop():
global agent_instance, browser_running
try:
if agent_instance:
web_interface.log('BROWSER', '正在关闭浏览器...')
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(agent_instance.stop())
loop.close()
agent_instance = None
browser_running = False
web_interface.update_browser_status(False, '浏览器已停止')
web_interface.log('BROWSER', '浏览器已关闭', level='success')
web_interface.print_system('浏览器已停止')
except Exception as e:
web_interface.print_error(f'关闭浏览器出错: {str(e)}')
web_interface.log('ERROR', str(e), level='error')
threading.Thread(target=stop, daemon=True).start()
@socketio.on('user_input')
def handle_user_input(data):
"""处理用户输入"""
global browser_running
if not browser_running or agent_instance is None:
web_interface.print_error('浏览器未启动，请先启动浏览器')
return
content = data.get('content', '').strip()
if not content:
return
# 显示用户消息
web_interface.print_user(content)
web_interface.emit('typing', {})
# 在后台线程中处理
def process():
asyncio.run(process_user_input(content))
threading.Thread(target=process, daemon=True).start()
async def process_user_input(user_input: str):
"""处理用户输入（异步）"""
try:
# 发送用户输入给 DeepSeek
await agent_instance.send_message(user_input)
# 等待并获取响应
response = await agent_instance.wait_for_response()
if response:
# 解析 ACTION
action = agent_instance._parse_action(response)
# 显示完整的原始响应
web_interface.print_ai(response, raw=response)
# 记录日志
web_interface.log('AGENT', f'收到响应: {len(response)} 字符')
if action:
web_interface.log('ACTION', f'检测到 ACTION: {action.get("type")}')
# 执行命令
result = await agent_instance.engine.execute(
action.get('type', ''),
action.get('cmd', ''),
action.get('cwd', ''),
action.get('content', ''),
agent=agent_instance
)
# 显示执行结果
output = result.get('output', '')
web_interface.log('RESULT', f'执行结果: {output[:200]}...' if len(output) > 200 else f'执行结果: {output}')
# 发送结果回 DeepSeek
await agent_instance.send_result(output)
else:
web_interface.print_error('未收到 DeepSeek 的响应')
except Exception as e:
web_interface.print_error(f'处理出错: {str(e)}')
import traceback
web_interface.log('ERROR', traceback.format_exc(), level='error')
@socketio.on('get_terminal_output')
def handle_get_terminal():
"""获取终端输出"""
if agent_instance and agent_instance.engine:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(
agent_instance.engine.execute('get_terminal_output', '', '', '', agent=agent_instance)
)
loop.close()
web_interface.emit('terminal_output', {'content': result.get('output', '')})
@socketio.on('clear_terminal')
def handle_clear_terminal():
"""清空终端"""
if agent_instance and agent_instance.engine:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(
agent_instance.engine.execute('clear_terminal', '', '', '', agent=agent_instance)
)
loop.close()
web_interface.print_system('终端输出已清空')
def main():
print("=" * 60)
print("DeepSeek Agent Web 控制台")
print("=" * 60)
print(f"请打开浏览器访问: http://localhost:5000")
print("=" * 60)
# 启动 Flask-SocketIO 服务器
socketio.run(app, host='0.0.0.0', port=5000, debug=False, allow_unsafe_werkzeug=True)
if __name__ == '__main__':
main()