From 898f18036149c425bff40253e4886640a92d9e54 Mon Sep 17 00:00:00 2001 From: admin Date: Fri, 17 Apr 2026 09:12:38 +0800 Subject: [PATCH] =?UTF-8?q?init:=20DeepSeek=20Agent=20Web=20=E6=8E=A7?= =?UTF-8?q?=E5=88=B6=E5=8F=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 6 + README.md | 28 + detect_generating.py | 85 ++++ web_server.py | 869 +++++++++++++++++++++++++++++++ yuanbao_agent_chat.py | 1133 +++++++++++++++++++++++++++++++++++++++++ 5 files changed, 2121 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 detect_generating.py create mode 100644 web_server.py create mode 100644 yuanbao_agent_chat.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..826e618 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.pyc +*.pyo +*.pyd +agent_debug.log +.workbuddy/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..719c0e4 --- /dev/null +++ b/README.md @@ -0,0 +1,28 @@ +# 元宝全自动代理 + +## 安装（conda 版本） + +```bash +# 创建 conda 环境 +conda create -n yuanbao-agent python=3.12 -y + +# 激活环境 +conda activate yuanbao-agent + +# 安装 playwright（包含浏览器） +pip install playwright +python -m playwright install chrome +``` + +## 运行 + +```bash +# 确保在 yuanbao-agent 环境中 +conda activate yuanbao-agent + +# 运行 MVP v1（显式指令） +python yuanbao_agent_mvp.py + +# 运行 v2 智能版（推荐） +python yuanbao_agent_v2.py +``` diff --git a/detect_generating.py b/detect_generating.py new file mode 100644 index 0000000..7c1c1b2 --- /dev/null +++ b/detect_generating.py @@ -0,0 +1,85 @@ +""" +探测元宝页面的生成状态特征 +运行后，请在元宝页面发送一条消息，观察输出 +""" +import asyncio +from playwright.async_api import async_playwright +from pathlib import Path + +YUANBAO_URL = "https://yuanbao.tencent.com/chat" + +async def detect(): + async with async_playwright() as p: + user_data = str(Path.home() / ".yuanbao_chat_profile") + context = await p.chromium.launch_persistent_context( + user_data, channel="chrome", headless=False, + args=["--start-maximized"], no_viewport=True + ) + + page = context.pages[0] if context.pages else await context.new_page() + await page.goto(YUANBAO_URL) + + print("=" * 60) + print("元宝生成状态探测器") + print("=" * 60) + print("\n1. 请在元宝页面登录并创建新对话") + print("2. 在元宝页面发送一条消息（比如'你好'）") + print("3. 观察生成过程中和生成完成后的输出差异") + print("4. 按 Ctrl+C 退出") + print("=" * 60) + + input("\n准备好后按回车开始探测...") + + # 持续探测 + prev_state = None + counter = 0 + + while True: + try: + # 获取最后一条AI消息的完整信息 + result = await page.evaluate(""" + () => { + const aiMessages = document.querySelectorAll( + '.agent-chat__list__item--ai, [class*="agent-chat"][class*="ai"], [class*="message"][class*="ai"]' + ); + if (aiMessages.length === 0) return null; + + const lastMsg = aiMessages[aiMessages.length - 1]; + const textEl = lastMsg.querySelector('.agent-chat__speech-text, [class*="text"], [class*="content"]'); + + return { + className: lastMsg.className, + innerHTML: lastMsg.innerHTML.substring(0, 500), + textContent: textEl ? textEl.innerText.substring(0, 200) : '', + hasCursor: lastMsg.innerHTML.includes('cursor') || lastMsg.innerHTML.includes('blink'), + childElements: Array.from(lastMsg.children).map(c => c.tagName + '.' + c.className).slice(0, 5) + }; + } + """) + + if result: + state_hash = f"{result['className']}_{result['hasCursor']}" + + if state_hash != prev_state: + prev_state = state_hash + counter += 1 + print(f"\n--- 探测 #{counter} ---") + print(f"Class: {result['className']}") + print(f"Has Cursor: {result['hasCursor']}") + print(f"Child Elements: {result['childElements']}") + print(f"Text Preview: {result['textContent'][:100]}...") + print(f"HTML Snippet: {result['innerHTML'][:200]}...") + + await asyncio.sleep(0.5) + + except KeyboardInterrupt: + break + except Exception as e: + print(f"Error: {e}") + await asyncio.sleep(1) + + await context.close() + print("\n探测结束") + +if __name__ == "__main__": + asyncio.run(detect()) diff --git a/web_server.py b/web_server.py new file mode 100644 index 0000000..62287b7 --- /dev/null +++ b/web_server.py @@ -0,0 +1,869 @@ +""" +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 + +# 导入 yuanbao_agent_chat 的核心类 +from yuanbao_agent_chat import SoloAgent, ExecutionEngine + +app = Flask(__name__) +app.config['SECRET_KEY'] = 'yuanbao-agent-secret' +socketio = SocketIO(app, cors_allowed_origins="*", async_mode='threading') + +# 全局状态 +agent_instance = None +browser_running = False +message_queue = queue.Queue() + +# HTML 模板 +HTML_TEMPLATE = ''' + + + + + + Yuanbao Agent - Web 控制台 + + + + +
+

🤖 Yuanbao Agent Web 控制台

+
+
+ + 浏览器未启动 +
+
+ + +
+
+
+ +
+ + + +
+ +
+
+
+
+ DeepSeek 正在思考... +
+
+
+ + +
+
+
+ +
+
+
运行日志
+
命令终端
+
+
+
+ +
+
+
+ + + + +''' + + +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("Yuanbao 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() diff --git a/yuanbao_agent_chat.py b/yuanbao_agent_chat.py new file mode 100644 index 0000000..af4b00b --- /dev/null +++ b/yuanbao_agent_chat.py @@ -0,0 +1,1133 @@ +""" +DeepSeek Agent - 分离式聊天界面 + +用户在这个终端界面与 Agent 对话 +Agent 后台与 DeepSeek 交互，自动处理 ACTION 标记 + +使用方式: +1. 启动此脚本 +2. 在终端直接输入任务 +3. Agent 自动协调 DeepSeek 完成多轮操作 +4. 结果返回给用户 +""" + +import asyncio +import json +import re +import subprocess +import time +import os +import sys +import threading +import queue +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional, Any, Callable +from enum import Enum, auto + +from playwright.async_api import async_playwright, Page, BrowserContext + +DEEPSEEK_URL = "https://chat.deepseek.com" + + +SOLO_SYSTEM_PROMPT = """【系统指令】你是一个命令执行代理。你的唯一职责是根据用户指令执行本地命令，并将执行结果返回给用户。 + +═══════════════════════════════════════════════════════ +【系统架构 - 重要！】 +═══════════════════════════════════════════════════════ +本系统采用【双终端架构】： +• 【DP对话终端】- 你当前所在的终端，用于和我（系统）对话 +• 【命令执行终端】- 独立的shell进程，专门用于执行所有命令 + +关键机制： +1. command/powershell 命令会发送到【命令执行终端】执行 +2. 命令执行是【非阻塞】的 - 命令扔进去立即返回，不等待完成 +3. 命令执行终端的输出会【持续累积】，随时可以获取 +4. 你需要用 get_terminal_output 主动查询命令执行结果 + +═══════════════════════════════════════════════════════ +【可用命令】 +═══════════════════════════════════════════════════════ +• command <命令> - 在命令执行终端执行命令（非阻塞） +• powershell <命令> - 在命令执行终端执行PowerShell（非阻塞） +• get_terminal_output - 获取命令执行终端的所有输出 +• send_input <内容> - 向命令执行终端发送输入（用于交互式程序） +• clear_terminal - 清空命令执行终端的输出记录 +• read_file <路径> - 读取文件内容 +• write_file <路径> - 写入文件（配合 << - 列出目录内容 +• search <关键词> - 搜索文件内容 +• complete - 标记任务完成 + +═══════════════════════════════════════════════════════ +【执行流程 - 必须遵守】 +═══════════════════════════════════════════════════════ +第1步：发送 command 执行命令 + ↓ +第2步：等待系统返回（命令已发送，不等待完成） + ↓ +第3步：发送 get_terminal_output 查询输出 + ↓ +第4步：根据输出判断： + - 命令已完成 → 继续下一步 + - 命令在运行中 → 等待后再次 get_terminal_output + - 命令在等待输入 → 用 send_input 发送输入 + - 命令出错 → 处理错误 + +═══════════════════════════════════════════════════════ +【响应格式 - 必须严格遵守】 +═══════════════════════════════════════════════════════ +每次响应必须包含两部分： + +1. ACTION 块（如需执行命令）： + <<>> + +2. 结束标记（必须！）： + <<>> + +═══════════════════════════════════════════════════════ +【重要规则】 +═══════════════════════════════════════════════════════ +1. 【禁止主动执行】初始化阶段严禁执行任何命令，必须等待用户明确指令 +2. 【必须加结束标记】每次回复的最后一行必须是 <<>> +3. 【非阻塞执行】command/powershell 不等待命令完成，立即返回 +4. 【必须查询输出】执行命令后，必须用 get_terminal_output 查看结果 +5. 【交互式程序】如果程序等待输入，用 send_input 发送输入，不要假设程序已结束 +6. 【文件编码】write_file 配合 <<>> +<<>> + +系统返回：命令已发送... + +你（第2轮）： +<<>> +<<>> + +系统返回：命令终端输出...（显示脚本输出） + +你（第3轮，如果程序在等待输入）： +<<>> +<<>> + +═══════════════════════════════════════════════════════ +【示例 - 写入文件（必须用代码围栏）】 +═══════════════════════════════════════════════════════ +<<>> +<<>> +<<>> + +═══════════════════════════════════════════════════════ +【示例 - 简单命令】 +═══════════════════════════════════════════════════════ +用户：查看当前目录 + +你： +<<>> +<<>> + +任务完成时： +<<>> +<<>> +""" + + +class ExecutionEngine: + """本地命令执行引擎 - 双终端架构""" + + def __init__(self): + # 命令执行终端（独立 shell，非阻塞执行命令） + self.cmd_terminal_proc: Optional[asyncio.subprocess.Process] = None + self.cmd_terminal_output: str = "" + self.cmd_terminal_lock = asyncio.Lock() + self.last_cmd: str = "" + + async def _ensure_cmd_terminal(self) -> asyncio.subprocess.Process: + """确保命令执行终端已启动""" + if self.cmd_terminal_proc is None or self.cmd_terminal_proc.returncode is not None: + if sys.platform == "win32": + self.cmd_terminal_proc = await asyncio.create_subprocess_exec( + "powershell", "-NoLogo", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + stdin=asyncio.subprocess.PIPE, + creationflags=subprocess.CREATE_NO_WINDOW + ) + # 启动后立即设置 UTF-8 编码 + await asyncio.sleep(0.2) + self.cmd_terminal_proc.stdin.write(b"[Console]::OutputEncoding = [System.Text.Encoding]::UTF8\n") + await self.cmd_terminal_proc.stdin.drain() + else: + self.cmd_terminal_proc = await asyncio.create_subprocess_exec( + "bash", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + stdin=asyncio.subprocess.PIPE + ) + # 启动后台任务持续读取输出 + asyncio.create_task(self._read_cmd_terminal_output()) + print(f"[系统] 命令执行终端已启动 (PID: {self.cmd_terminal_proc.pid})") + return self.cmd_terminal_proc + + async def _read_cmd_terminal_output(self): + """后台任务：持续读取命令终端的输出""" + while self.cmd_terminal_proc and self.cmd_terminal_proc.returncode is None: + try: + line = await asyncio.wait_for( + self.cmd_terminal_proc.stdout.readline(), + timeout=0.1 + ) + if line: + async with self.cmd_terminal_lock: + decoded = line.decode('utf-8', errors='replace') + self.cmd_terminal_output += decoded + except asyncio.TimeoutError: + continue + except Exception: + break + + async def execute(self, action_type: str, command: str, working_dir: str = "", content: str = "", agent=None) -> dict: + cwd = Path(working_dir).expanduser() if working_dir else Path.home() + + handlers = { + "command": self._exec_in_cmd_terminal, + "powershell": self._exec_in_cmd_terminal, + "read_file": self._read_file, + "write_file": self._write_file, + "list_dir": self._list_dir, + "search": self._search, + "get_terminal_output": self._get_terminal_output, + "send_input": self._send_input_to_terminal, + "clear_terminal": self._clear_terminal_output, + } + + handler = handlers.get(action_type) + if not handler: + return {"success": False, "output": f"Unknown action type: {action_type}"} + + try: + if action_type == "write_file": + # write_file 特殊处理，传入 content 和 agent + if asyncio.iscoroutinefunction(handler): + result = await handler(command, cwd, content, agent) + else: + result = handler(command, cwd, content, agent) + else: + if asyncio.iscoroutinefunction(handler): + result = await handler(command, cwd) + else: + result = handler(command, cwd) + return {"success": True, "output": result} + except Exception as e: + return {"success": False, "output": f"Error: {e}"} + + async def _exec_in_cmd_terminal(self, cmd: str, cwd: Path) -> str: + """在命令执行终端中执行命令（非阻塞，立即返回）""" + proc = await self._ensure_cmd_terminal() + + # 记录执行前的输出位置 + async with self.cmd_terminal_lock: + output_before_len = len(self.cmd_terminal_output) + + # 发送命令到命令终端 + self.last_cmd = cmd + if sys.platform == "win32": + # 设置 Python 环境变量强制 UTF-8 输出 + if cmd.strip().startswith("python"): + full_cmd = f"cd '{cwd}'; $env:PYTHONIOENCODING='utf-8'; {cmd}\n" + else: + full_cmd = f"cd '{cwd}'; {cmd}\n" + else: + full_cmd = f"cd '{cwd}' && {cmd}\n" + + proc.stdin.write(full_cmd.encode()) + await proc.stdin.drain() + + # 等待一段时间让命令开始执行（不等待完成） + await asyncio.sleep(0.5) + + # 获取新产生的输出 + async with self.cmd_terminal_lock: + current_output = self.cmd_terminal_output[output_before_len:] + + return f"[已执行] {cmd}\n\n命令终端当前输出:\n```\n{current_output[-2000:] if len(current_output) > 2000 else current_output}\n```\n\n提示: 使用 'get_terminal_output' 查看最新输出" + + async def _get_terminal_output(self, _: str, cwd: Path) -> str: + """获取命令执行终端的当前所有输出""" + async with self.cmd_terminal_lock: + output = self.cmd_terminal_output + + if len(output) > 5000: + return f"命令终端输出 (共 {len(output)} 字符，显示最后5000字符):\n```\n...\n{output[-5000:]}\n```" + else: + return f"命令终端输出 (共 {len(output)} 字符):\n```\n{output}\n```" + + async def _send_input_to_terminal(self, input_text: str, cwd: Path) -> str: + """向命令执行终端发送输入""" + if not self.cmd_terminal_proc or self.cmd_terminal_proc.returncode is not None: + return "错误: 命令终端未启动" + + self.cmd_terminal_proc.stdin.write(input_text.encode() + b'\n') + await self.cmd_terminal_proc.stdin.drain() + + # 等待一下让程序处理输入 + await asyncio.sleep(0.3) + + # 获取新输出 + async with self.cmd_terminal_lock: + output = self.cmd_terminal_output + + return f"[已发送输入] {input_text}\n\n终端最新输出:\n```\n{output[-1500:] if len(output) > 1500 else output}\n```" + + async def _clear_terminal_output(self, _: str, cwd: Path) -> str: + """清空命令终端输出记录""" + async with self.cmd_terminal_lock: + old_len = len(self.cmd_terminal_output) + self.cmd_terminal_output = "" + return f"已清空命令终端输出记录 (清除了 {old_len} 字符)" + + def _read_file(self, path: str, cwd: Path) -> str: + target = cwd / path if not Path(path).is_absolute() else Path(path) + if not target.exists(): + return f"File not found: {target}" + try: + return target.read_text(encoding='utf-8', errors='replace')[:5000] + except Exception as e: + return f"Read error: {e}" + + def _write_file(self, cmd: str, cwd: Path, content: str = "", agent=None) -> str: + # 新格式: path 直接传入，content 从 action 对象获取 + # 兼容旧格式: path::content + if "::" in cmd and not content: + path, content = cmd.split("::", 1) + else: + path = cmd + + if agent: + agent.log("WRITE_FILE", f"Path: {path}, Raw content length: {len(content)}") + agent.log("WRITE_FILE", f"Raw content preview: {content[:300].replace(chr(10), '\\n')}") + + # 处理代码围栏：去掉 ``` 或 ```python 等标记 + content = content.strip() + + # 更健壮的围栏处理：找到第一个 ``` 和最后一个 ``` + if content.startswith('```'): + # 找到第一个换行符（跳过 ```python 这一行） + first_newline = content.find('\n') + if first_newline != -1: + # 去掉开头的 ``` 或 ```python 行 + content = content[first_newline + 1:] + + # 去掉结尾的 ``` + if content.rstrip().endswith('```'): + last_backticks = content.rstrip().rfind('```') + content = content[:last_backticks].rstrip() + + if agent: + agent.log("WRITE_FILE", f"After fence removal: {len(content)} chars") + agent.log("WRITE_FILE", f"Content preview: {content[:300].replace(chr(10), '\\n')}") + + # 检查是否还有残留的 ``` + if '```' in content: + if agent: + agent.log("WRITE_FILE", f"WARNING: Still contains ``` after removal!") + + # 检查缩进 + lines_with_indent = [i for i, line in enumerate(content.split('\n')) if line.startswith(' ') or line.startswith('\t')] + if agent: + agent.log("WRITE_FILE", f"Lines with indentation: {lines_with_indent}") + agent.log("WRITE_FILE", f"Final content: {content[:300].replace(chr(10), '\\n')}") + + target = cwd / path if not Path(path).is_absolute() else Path(path) + try: + target.write_text(content, encoding='utf-8') + result = f"Written to {target} ({len(content)} chars)" + if agent: + agent.log("WRITE_FILE", f"Success: {result}") + return result + except Exception as e: + if agent: + agent.log("WRITE_FILE", f"Error: {e}") + return f"Write error: {e}" + + def _list_dir(self, path: str, cwd: Path) -> str: + target = cwd / path if not Path(path).is_absolute() else Path(path) + if not target.exists(): + return f"Directory not found: {target}" + try: + items = [] + for item in target.iterdir(): + item_type = "📁" if item.is_dir() else "📄" + size = "" + if item.is_file(): + try: + size_bytes = item.stat().st_size + if size_bytes < 1024: + size = f" ({size_bytes}B)" + elif size_bytes < 1024*1024: + size = f" ({size_bytes/1024:.1f}KB)" + else: + size = f" ({size_bytes/(1024*1024):.1f}MB)" + except: + pass + items.append(f"{item_type} {item.name}{size}") + return "\n".join(items) if items else "(empty directory)" + except Exception as e: + return f"List error: {e}" + + def _search(self, pattern: str, cwd: Path) -> str: + results = [] + try: + for item in cwd.rglob("*"): + if item.is_file() and pattern.lower() in item.name.lower(): + results.append(str(item.relative_to(cwd))) + if len(results) >= 20: + results.append("... (more results)") + break + return "\n".join(results) if results else "No matches found" + except Exception as e: + return f"Search error: {e}" + + +class DeepSeekController: + """DeepSeek 页面控制器""" + + def __init__(self, page: Page): + self.page = page + + async def send(self, message: str) -> bool: + """发送消息到 DeepSeek""" + try: + # 查找输入框 + input_selectors = [ + 'textarea[placeholder*="输入"]', + 'textarea[class*="input"]', + '[class*="input"] textarea', + 'div[contenteditable="true"]', + 'textarea', + ] + + input_box = None + for selector in input_selectors: + try: + el = self.page.locator(selector).first + if await el.count() > 0 and await el.is_visible(): + input_box = el + break + except: + continue + + if not input_box: + print("[Error] 找不到输入框") + return False + + await input_box.fill(message) + await asyncio.sleep(0.3) + + # 点击发送按钮 + send_btn = self.page.locator("button[class*='send']").last + if await send_btn.count() > 0 and await send_btn.is_visible(): + await send_btn.click() + else: + await input_box.press("Enter") + + return True + except Exception as e: + print(f"[Error] 发送失败: {e}") + return False + + async def get_latest_ai_message(self, agent: 'SoloAgent' = None) -> Optional[str]: + """获取最新 AI 消息 - 适配 DeepSeek DOM 结构""" + try: + text = await self.page.evaluate(""" + () => { + // 辅助函数：提取消息内容，正确处理代码块 + function extractMessageContent(msg) { + // 找到 markdown 容器 + const markdownContent = msg.querySelector('[class*="markdown"], [class*="ds-markdown"]'); + if (!markdownContent) { + return msg.textContent.trim(); + } + + let result = ''; + + // 遍历所有子元素 + const children = markdownContent.children; + for (const child of children) { + // 代码块容器 + if (child.tagName === 'DIV' && child.querySelector('pre')) { + const pre = child.querySelector('pre'); + const code = pre.querySelector('code'); + // 只取 code 或 pre 的 textContent，忽略按钮等 UI + const codeText = code ? code.textContent : pre.textContent; + result += '\\n```python\\n' + codeText + '\\n```\\n'; + } + // 普通段落 + else if (child.tagName === 'P') { + result += child.textContent + '\\n'; + } + // 其他元素直接取文本 + else { + result += child.textContent + '\\n'; + } + } + + return result.trim(); + } + + // 找所有消息容器 + const messageContainers = [ + ...document.querySelectorAll('[class*="ds-message"]'), + ...document.querySelectorAll('[class*="message"]'), + ]; + + if (messageContainers.length === 0) return null; + + // 遍历消息，找最后一个包含结束标记的 AI 消息 + for (let i = messageContainers.length - 1; i >= 0; i--) { + const msg = messageContainers[i]; + const content = extractMessageContent(msg); + + // 找包含结束标记的消息 + if (content.includes('<<>>')) { + return content; + } + } + + // 兜底：返回最后一个消息 + const lastMsg = messageContainers[messageContainers.length - 1]; + return extractMessageContent(lastMsg); + } + """) + + if agent and text: + agent.log("MSG_EXTRACT", f"Extracted: {text[:300].replace(chr(10), '\\n')}") + + return text + except Exception as e: + print(f"[Error] 获取消息失败: {e}") + return None + + +class ChatInterface: + """用户聊天界面 - 在终端与用户交互""" + + def __init__(self): + self.message_queue = queue.Queue() + self.response_queue = queue.Queue() + self.is_running = False + self.task_in_progress = False + + def print_banner(self): + """打印欢迎信息""" + print("\n" + "=" * 60) + print(" [DeepSeek Agent] - 分离式聊天界面") + print("=" * 60) + print("\n[使用说明]:") + print(" 1. 在此窗口直接输入任务") + print(" 2. Agent 自动后台协调 DeepSeek 完成") + print(" 3. 多轮操作完成后返回结果") + print(" 4. DeepSeek 窗口保持自动化交互，无需手动操作") + print("\n[示例任务]:") + print(" - 查看桌面有什么文件") + print(" - 统计当前目录的Python文件数量") + print(" - 查找大于100MB的文件") + print("\n[特殊命令]:") + print(" /quit - 退出程序") + print(" /clear - 清屏") + print("=" * 60 + "\n") + + def user_input_loop(self): + """用户输入循环 - 在单独线程运行""" + while self.is_running: + try: + # 显示提示符 + if not self.task_in_progress: + user_input = input("\n[你]: ") + + if user_input.strip(): + if user_input.strip().lower() == '/quit': + self.is_running = False + break + elif user_input.strip().lower() == '/clear': + os.system('cls' if sys.platform == 'win32' else 'clear') + self.print_banner() + continue + + # 将用户输入放入队列 + self.message_queue.put(user_input.strip()) + self.task_in_progress = True + print("\n[正在处理...]") + else: + time.sleep(0.1) + + except EOFError: + break + except KeyboardInterrupt: + self.is_running = False + break + + def display_result(self, result: str): + """显示任务结果""" + print(f"\n[DeepSeek]: {result}\n") + self.task_in_progress = False + + def display_progress(self, msg: str): + """显示进度信息""" + print(f" {msg}") + + +END_MARKER = "<<>>" + + +class SoloAgent: + """主 Agent - 协调用户界面和 DeepSeek""" + + def __init__(self): + self.page: Optional[Page] = None + self.controller: Optional[DeepSeekController] = None + self.engine = ExecutionEngine() + self.chat = ChatInterface() + self.last_ai_msg = "" + self.log_file = Path("agent_debug.log") + self._init_log() + self._playwright = None + self._browser = None + self._context = None + self._initialized = False + + async def start(self): + """启动浏览器和 DeepSeek 连接""" + if self._initialized: + return + + self._playwright = await async_playwright().start() + + # 使用持久化上下文，保存登录状态 + user_data_dir = Path.home() / ".deepseek_chat_profile" + user_data_dir.mkdir(exist_ok=True) + + try: + # 尝试启动持久化上下文 + self._context = await self._playwright.chromium.launch_persistent_context( + str(user_data_dir), + channel="chrome", + headless=False, + args=["--start-maximized"], + no_viewport=True + ) + print("[使用持久化会话，保留登录状态]") + except Exception as e: + # 如果失败，回退到普通模式 + print(f"[!] 持久化会话启动失败: {e}") + print("[!] 使用普通模式（需要重新登录）") + self._browser = await self._playwright.chromium.launch( + channel="chrome", + headless=False, + args=["--start-maximized"] + ) + self._context = await self._browser.new_context(no_viewport=True) + + self.page = self._context.pages[0] if self._context.pages else await self._context.new_page() + await self.page.goto(DEEPSEEK_URL) + + print(f"[已打开 DeepSeek]: {DEEPSEEK_URL}") + print("[!] 请先在浏览器中登录 DeepSeek 并创建新对话") + + # 等待用户登录（阻塞等待） + input("登录完成后按回车继续...") + + self.controller = DeepSeekController(self.page) + + # 发送系统提示词 + print("[正在初始化系统提示词]...") + await self.controller.send(f"[系统初始化]\n\n{SOLO_SYSTEM_PROMPT}") + print("[系统初始化完成]") + + self._initialized = True + + async def stop(self): + """停止浏览器""" + if self._context: + await self._context.close() + self._context = None + if self._browser: + await self._browser.close() + self._browser = None + if self._playwright: + await self._playwright.stop() + self._playwright = None + self._initialized = False + self.page = None + self.controller = None + + async def send_message(self, message: str): + """发送消息给 DeepSeek""" + if not self.controller: + raise RuntimeError("Agent 未启动") + await self.controller.send(f"[用户任务]\n{message}\n\n请开始执行。") + self.last_ai_msg = "" + + async def wait_for_response(self, timeout: int = 120) -> Optional[str]: + """等待 DeepSeek 响应""" + if not self.controller: + raise RuntimeError("Agent 未启动") + + start_time = time.time() + last_msg = self.last_ai_msg + + while time.time() - start_time < timeout: + current_msg = await self.controller.get_latest_ai_message(self) + if current_msg and current_msg != last_msg: + self.last_ai_msg = current_msg + return current_msg + await asyncio.sleep(0.5) + + return None + + async def send_result(self, result: str): + """发送执行结果给 DeepSeek""" + if not self.controller: + raise RuntimeError("Agent 未启动") + await self.controller.send(f"[执行结果]\n{result}\n\n请继续下一步，或标记任务完成。") + + def _init_log(self): + """初始化日志文件 - 每次运行清空旧日志""" + timestamp = time.strftime("%Y-%m-%d %H:%M:%S") + # 清空旧日志，只保留当前运行的记录 + self.log_file.write_text(f"=== Agent Debug Log - {timestamp} ===\n\n", encoding='utf-8') + self._log_buffer = [] # 内存中的日志缓冲区 + + def read_log(self, lines: int = 50) -> str: + """读取最近 N 行日志""" + try: + if self.log_file.exists(): + content = self.log_file.read_text(encoding='utf-8') + all_lines = content.split('\n') + return '\n'.join(all_lines[-lines:]) + except: + pass + return "(日志为空)" + + def log(self, category: str, message: str): + """写入日志""" + timestamp = time.strftime("%H:%M:%S") + log_line = f"[{timestamp}] [{category}] {message}\n" + with open(self.log_file, 'a', encoding='utf-8') as f: + f.write(log_line) + # 同时保存到内存缓冲区 + self._log_buffer.append((category, message)) + # 只保留最近100条 + if len(self._log_buffer) > 100: + self._log_buffer = self._log_buffer[-100:] + + def analyze_last_operation(self) -> str: + """分析最近的操作日志，找出问题""" + analysis = [] + + # 查找最近的 WRITE_FILE 操作 + write_file_logs = [(i, cat, msg) for i, (cat, msg) in enumerate(self._log_buffer) if cat == "WRITE_FILE"] + + if write_file_logs: + analysis.append("=== 文件写入分析 ===") + for idx, cat, msg in write_file_logs[-3:]: # 最近3次 + analysis.append(f" {msg}") + + # 检查缩进问题 + if "Lines with indentation" in msg: + if "[]" in msg or " Lines with indentation: 0" in msg: + analysis.append(" ⚠️ 警告：没有检测到缩进！") + + # 查找 CONTENT 提取 + content_logs = [(i, cat, msg) for i, (cat, msg) in enumerate(self._log_buffer) if cat == "PARSE" and "CONTENT" in msg] + if content_logs: + analysis.append("\n=== CONTENT 提取分析 ===") + for idx, cat, msg in content_logs[-3:]: + analysis.append(f" {msg}") + + # 查找原始消息 + raw_logs = [(i, cat, msg) for i, (cat, msg) in enumerate(self._log_buffer) if cat == "RAW_MSG"] + if raw_logs: + analysis.append("\n=== 原始消息分析 ===") + for idx, cat, msg in raw_logs[-2:]: + # 检查是否包含 ``` + if "```" in msg: + analysis.append(f" ✓ 检测到代码围栏标记") + else: + analysis.append(f" ✗ 未检测到代码围栏标记") + + # 检查缩进字符 + if "\\n " in msg or "\\n\\t" in msg: + analysis.append(f" ✓ 原始消息包含缩进字符") + else: + analysis.append(f" ⚠️ 原始消息可能没有缩进") + + return "\n".join(analysis) if analysis else "(暂无分析数据)" + + + async def run(self): + """主运行循环""" + self.chat.print_banner() + + async with async_playwright() as p: + # 使用持久化上下文，保存登录状态 + from pathlib import Path + user_data_dir = Path.home() / ".deepseek_chat_profile" + user_data_dir.mkdir(exist_ok=True) + + try: + # 尝试启动持久化上下文 + context = await p.chromium.launch_persistent_context( + str(user_data_dir), + channel="chrome", + headless=False, + args=["--start-maximized"], + no_viewport=True + ) + print("[使用持久化会话，保留登录状态]") + except Exception as e: + # 如果失败，回退到普通模式 + print(f"[!] 持久化会话启动失败: {e}") + print("[!] 使用普通模式（需要重新登录）") + browser = await p.chromium.launch( + channel="chrome", + headless=False, + args=["--start-maximized"] + ) + context = await browser.new_context(no_viewport=True) + + self.page = context.pages[0] if context.pages else await context.new_page() + await self.page.goto(DEEPSEEK_URL) + + print(f"[已打开 DeepSeek]: {DEEPSEEK_URL}") + print("[!] 请先在 DeepSeek 网页中登录并创建新对话\n") + + input("登录完成后按回车继续...") + + self.controller = DeepSeekController(self.page) + + # 发送系统提示词 + system_prompt = SOLO_SYSTEM_PROMPT + + print("[正在初始化系统提示词]...") + await self.controller.send(f"[系统初始化]\n\n{system_prompt}") + print("[系统初始化完成]\n") + + # 启动用户输入线程 + self.chat.is_running = True + input_thread = threading.Thread(target=self.chat.user_input_loop, daemon=True) + input_thread.start() + + # 主处理循环 + await self._main_loop() + + async def _main_loop(self): + """主处理循环""" + while self.chat.is_running: + try: + # 检查是否有用户输入 + if not self.chat.message_queue.empty(): + user_task = self.chat.message_queue.get() + await self._process_task(user_task) + + await asyncio.sleep(0.1) + + except Exception as e: + print(f"\n❌ 错误: {e}\n") + self.chat.task_in_progress = False + + async def _process_task(self, user_task: str): + """处理用户任务 - 与 DeepSeek 多轮交互""" + print(f"\n[任务]: {user_task}") + print("-" * 50) + + # 发送用户任务给 DeepSeek + await self.controller.send(f"[用户任务]\n{user_task}\n\n请开始执行。") + + # 初始化消息追踪 + await asyncio.sleep(2) + self.last_ai_msg = await self.controller.get_latest_ai_message(self) or "" + + round_count = 0 + + while True: + round_count += 1 + + # 等待 DeepSeek 回复 + await asyncio.sleep(2) + current_msg = await self._wait_for_new_message() + + # 记录获取到的消息 + if current_msg: + self.log("MSG_RECEIVED", f"Length: {len(current_msg)}, Preview: {current_msg[:200].replace(chr(10), '\\n')}") + + if not current_msg: + print("[等待回复超时]") + break + + print(f"\n[DeepSeek 回复 - 轮次 {round_count}]:") + # 显示清理后的消息（去掉 ACTION 标记） + clean_msg = self._clean_action_markers(current_msg) + if clean_msg.strip(): + print(f" {clean_msg[:200]}...") + + # 解析 ACTION + action = self._parse_action(current_msg, self) + + if action is None and "<<>>" in current_msg: + # 有 ACTION 标记但解析失败，返回错误让 DeepSeek 重新输出 + print("\n[!] ACTION 解析失败，返回错误给 DeepSeek") + + # 提取原始 ACTION 内容给 DeepSeek 看 + import re + action_match = re.search(r'<<>>', current_msg, re.DOTALL) + raw_action = action_match.group(1) if action_match else "无法提取" + + # 获取具体的解析错误信息 + import json + try: + json.loads(raw_action) + parse_error = "未知错误" + except json.JSONDecodeError as e: + parse_error = str(e) + + error_msg = f"""[执行结果] +错误: ACTION 格式解析失败 + +解析错误：{parse_error} + +请重新输出正确的 ACTION。 + +本次回复请以 {END_MARKER} 结尾""" + + await asyncio.sleep(1) + await self.controller.send(error_msg) + + # 等待 DeepSeek 重新回复 + await asyncio.sleep(3) + self.last_ai_msg = "" + continue + + if action: + action_type = action.get('type', '') + + if action_type == 'complete': + # 任务完成 + explanation = action.get('explanation', '任务已完成') + print(f"\n[任务完成!]") + self.chat.display_result(explanation) + return + + # 执行命令 + cmd = action.get('cmd', '') + path = action.get('path', '') + content = action.get('content', '') + explanation = action.get('explanation', '') + + # write_file 新格式：用 path 而不是 cmd + if action_type == 'write_file' and path: + cmd = path + print(f"\n[执行]: [{action_type}] {path}") + else: + print(f"\n[执行]: [{action_type}] {cmd}") + + if explanation: + print(f" 说明: {explanation}") + + result = await self.engine.execute(action_type, cmd, "", content, self) + + # 显示执行结果摘要 + output_preview = result['output'][:200].replace('\n', ' ') + status = "[OK]" if result['success'] else "[ERR]" + print(f" {status} 结果: {output_preview}...") + + # 如果是 write_file 操作，自动分析问题 + if action_type == "write_file": + print("\n" + "="*50) + print("[自动诊断]") + print(self.analyze_last_operation()) + print("="*50) + + # 发送结果回 DeepSeek（完整输出，不截断） + display_cmd = path if action_type == 'write_file' and path else cmd + result_msg = f"""[执行结果] +操作: {action_type} +命令: {display_cmd} +成功: {result['success']} + +输出: +``` +{result['output']} +``` + +请继续下一步或完成任务。""" + + await asyncio.sleep(1) + await self.controller.send(result_msg) + + # 等待一段时间让 DeepSeek 开始生成回复 + print(" [等待 DeepSeek 继续...]") + await asyncio.sleep(3) + + # 重置 last_ai_msg，让 _wait_for_new_message 从头检测新消息 + self.last_ai_msg = "" + + else: + # 没有 ACTION 标记，可能是普通回复或任务完成 + if "完成" in current_msg or "done" in current_msg.lower(): + print(f"\n[任务完成!]") + self.chat.display_result(clean_msg) + return + else: + # 普通回复，显示给用户 + print(f"\n[DeepSeek]: {clean_msg}") + self.chat.display_result(clean_msg) + return + + async def _wait_for_new_message(self) -> Optional[str]: + """等待新消息完整输出 - 检测结束标记""" + CHECK_INTERVAL = 0.5 + + print(" [等待 DeepSeek 回复]...", end="", flush=True) + + while True: + # 获取当前消息内容 + current_msg = await self.controller.get_latest_ai_message() + + # 跳过已处理过的消息（完全相同的内容） + if not current_msg or current_msg == self.last_ai_msg: + await asyncio.sleep(CHECK_INTERVAL) + continue + + # 检查是否包含结束标记 + if END_MARKER in current_msg: + # 找到结束标记，消息完整 + print(f"\r [消息完成] ({len(current_msg)} 字符)") + # 去掉结束标记后返回 + clean_msg = current_msg.replace(END_MARKER, "").strip() + self.last_ai_msg = current_msg + return clean_msg + + # 还在生成中 + print(f"\r [生成中...] ({len(current_msg)} 字符)", end="", flush=True) + await asyncio.sleep(CHECK_INTERVAL) + + def _parse_action(self, text: str, agent: 'SoloAgent' = None) -> Optional[dict]: + """解析 ACTION 块和 CONTENT 块""" + if not text: + if agent: + agent.log("PARSE", "Empty text input") + return None + + # 记录原始消息（用于调试） + if agent: + agent.log("RAW_MSG", text[:500].replace('\n', '\\n')) + + # 调试：显示检测到的 ACTION + if "<<>>' + action_match = re.search(action_pattern, text, re.DOTALL) + + if not action_match: + # 备用解析 + if "<<>>" in text: + try: + start = text.find("<<>>") + json_content = text[start:end].strip() + print(f" [Debug] Fallback extraction: {json_content[:100]}...") + if agent: + agent.log("PARSE", f"Fallback extraction: {json_content[:200]}") + if json_content.startswith("{") and json_content.endswith("}"): + action_match = type('obj', (object,), {'group': lambda x: json_content})() + except Exception as e: + print(f" [Debug] Fallback error: {e}") + if agent: + agent.log("PARSE", f"Fallback error: {e}") + return None + else: + return None + + try: + json_str = action_match.group(1).strip() + print(f" [Debug] Extracted JSON: {json_str[:100]}...") + if agent: + agent.log("PARSE", f"JSON string: {json_str}") + action = json.loads(json_str) + + # 检查是否有配套的 CONTENT 块（用于 write_file） + if action.get("type") == "write_file": + content_pattern = r'<<>>' + content_match = re.search(content_pattern, text, re.DOTALL) + if content_match: + # 把 CONTENT 内容放到 action 里 + raw_content = content_match.group(1) + action["content"] = raw_content + print(f" [Debug] Extracted CONTENT: {len(action['content'])} chars") + if agent: + agent.log("PARSE", f"CONTENT extracted: {len(raw_content)} chars") + agent.log("PARSE", f"CONTENT preview: {raw_content[:200].replace(chr(10), '\\n')}") + elif "path" in action and "cmd" not in action: + # 新格式要求有 CONTENT 块但没有找到 + print(f" [Debug] Warning: write_file without CONTENT block") + if agent: + agent.log("PARSE", "Warning: write_file without CONTENT block") + action["content"] = "" + + return action + + except json.JSONDecodeError as e: + print(f" [Debug] JSON parse error: {e}") + if agent: + agent.log("PARSE", f"JSON error: {e}") + agent.log("PARSE", f"Problematic JSON: {json_str[:300] if 'json_str' in dir() else 'N/A'}") + return None + + def _clean_action_markers(self, text: str) -> str: + """清理 ACTION 标记，返回可读文本""" + # 移除 ACTION 块 + clean = re.sub(r'<<>>', '', text, flags=re.DOTALL) + # 清理多余空行 + clean = re.sub(r'\n{3,}', '\n\n', clean) + return clean.strip() + + +if __name__ == "__main__": + agent = SoloAgent() + try: + asyncio.run(agent.run()) + except KeyboardInterrupt: + print("\n\n[再见!]")