项目：ai-webot 作者：码上工坊 版本：0.1.0 日期：2026年2月 许可证：MIT License 项目地址1：https://gitee.com/icodewr/ai_webot 项目地址2：https://github.com/ICodeWR/ai_webot

AI Webot

网页端AI机器人聚合框架

项目起源

本项目源于实际用户需求。近期有位网友自行开发了一款软件，其生成的数据需要借助DeepSeek进行分析。起初我为其提供了基于DeepSeek API的调用方案，但在实际使用中发现了以下问题：

用户反馈的问题：

1. 成本因素 - API调用存在费用，对于高频使用场景不够经济
2. 响应延迟 - API响应速度相比Web版有明显延迟，影响分析效率
3. 操作繁琐 - 需要手动处理数据输入输出流程

用户在实际测试中发现，虽然API方案功能完整，但手动复制数据到DeepSeek Web版反而更符合其使用习惯——零成本、实时响应、交互直接。

基于这一需求，我们需要一个能够兼顾自动化与体验的解决方案，让用户既能保持Web版的优势，又能减少重复操作。

需求延展

另外一个场景（也是本人的需求）：多文件协同分析。特别是处理软件项目时，经常需要：

• 同时上传多个相关代码文件
• 希望能按目录结构批量上传
• 保持文件间关联性进行分析

项目目标

因此，本项目旨在开发一个工具，实现：

1. 直接对接DeepSeek Web版，保持Web版的实时响应速度
2. 支持数据自动导入和文件管理
3. 提供多文件和目录上传功能，满足工程分析需求

项目特性

多平台支持

核心功能

• 多AI聚合 - 一站式管理多个AI对话，无需频繁切换网站
• 文件处理 - 支持单文件/多文件上传，自动处理目录结构
• 实时交互 - 支持多轮对话
• 便捷集成 - 剪贴板一键复制，Markdown格式保存
• 登录状态 - 保存登录状态，下次免登录

本项目的学习价值

一个编程学习者的自动化工具实践 · 学中做 · 做中学

项目驱动 · 开源实践 · 实战学习 · 自造工具

本项目以 「为自己造工具」 为精神内核，驱动学习者通过亲手构建「AI机器人聚合」这一真实、完整的项目，贯通从学习、实践到产出个人作品的核心闭环。

学习哲学

• 好的学习是创造能用的东西 - 告别玩具代码，从第一行开始就构建真实可用的工具
• 真实的反馈来自自己每天使用 - 你将是自己产品的第一个也是最重要的用户
• 有效的成长来自解决自己的痛点 - 我的真实需求 = 最好的产品经理
• 纯粹的动力就是"我需要这个功能" - 为用而学，学以致用

「学做练用」深度循环

这是作者本人一贯坚持的学习理念

本项目技术学习路线

可扩展的多层项目架构

分层清晰的模块化架构

ai-webot/src/ai_webot/├── __init__.py              # 包导出定义├── api.py                   # 核心机器人工厂和注册表，学习设计模式├── cli.py                   # 命令行接口，学习命令行开发├── drivers/                 # 浏览器驱动层│   ├── __init__.py│   └── browser.py           # Playwright封装，学习浏览器自动化├── services/                # 服务层│   ├── __init__.py│   ├── config_service.py    # 配置管理，配置管理学习│   └── file_exceptions.py   # 异常处理，异常处理学习├── output/                  # 输出管理练习└── webot/                   # 机器人实现层    ├── __init__.py    ├── base/               # 抽象基类，学习抽象和继承    │   ├── __init__.py    │   └── web_bot.py      # 所有机器人的基类    ├── deepseek/           # DeepSeek实现，具体实现学习    │   ├── __init__.py    │   └── bot.py              ├── doubao/             # 豆包实现，平台适配练习    │   ├── __init__.py    │   └── bot.py    └── qianwen/            # 通义千问实现，平台适配练习        ├── __init__.py        └── bot.py

务实的技术栈选型

技术栈选型

基于上述需求，选择了以下技术组合：

# 项目依赖配置（pyproject.toml核心部分）[tool.poetry.dependencies]python = "^3.8"playwright = "^1.40.0"# 浏览器自动化框架pyyaml = "^6.0"# 配置文件支持

关键技术选型理由：

1. Playwright vs 其他方案：

• 对比Selenium：Playwright API更现代化，自动等待机制完善，对现代Web应用支持更好
• 对比直接API调用：各平台API限制严格，且频繁变更，浏览器自动化更稳定
• 实际验证：Playwright的跨浏览器支持和丰富的选择器策略，非常适合处理不同结构的网站

2. 异步编程的必要性：

• 浏览器操作是典型的I/O密集型任务，异步能大幅减少等待时间
• 支持多个机器人实例并行工作，实现真正的"聚合"效果
• Python 3.8+原生支持asyncio，生态成熟

3. 配置管理方案：

• 选择YAML作为配置文件格式，可读性好，支持注释
• 支持环境变量替换，方便不同环境部署
• 统一的配置服务层，便于维护和扩展

注：项目可支持两种配置格式

项目实现概要

核心设计模式实践

1. 工厂模式 + 注册表机制

在api.py中实现了智能的机器人发现和创建机制：

classBotRegistry:"""自动扫描配置文件目录，发现所有可用的机器人类型"""def__init__(self, config_dir: str = "configs"):        self.config_dir = Path(config_dir)        self._registry: Dict[str, Dict[str, Any]] = {}        self._config_service = ConfigService(config_dir)        self._scan_configs()  # 自动扫描配置文件def_scan_configs(self) -> None:"""扫描配置文件目录，发现可用机器人"""        config_files_info = self._get_config_files_info()# 按机器人类型和文件类型分组，优先使用YAML        bot_files: Dict[str, Dict[str, Any]] = {}for file_info in config_files_info:            bot_type = file_info["stem"]  # 如 "deepseek"# ... 解析配置，注册机器人

2. 抽象基类定义统一接口

webot/base/web_bot.py定义了所有机器人都必须实现的接口：

classWebBot(ABC):"""Web机器人抽象基类"""def__init__(self, config: BotConfig) -> None:        self.config = config        self.browser = BrowserDriver(config)  # 依赖注入浏览器驱动        self.is_logged_in = False        self.is_ready = False    @abstractmethoddefrequires_login(self) -> bool:"""判断该机器人是否需要登录"""pass    @abstractmethodasyncdeflogin(self) -> bool:"""执行登录操作"""pass    @abstractmethodasyncdefensure_ready(self) -> bool:"""确保机器人就绪可用"""pass# 公共方法提供默认实现asyncdefsend_message(self, message: str, files=None, dirs=None) -> str:"""发送消息的核心流程"""# 1. 确保机器人就绪ifnotawait self.ensure_ready():raise ValueError("机器人未就绪")# 2. 导航到聊天页面await self._ensure_chat_page()# 3. 处理文件上传if files:await self._upload_files(files)if dirs:await self._upload_directory(dirs)# 4. 输入并发送消息        input_selector = self.config.selectors.get("message_input")await self.browser.type(input_selector, message)await self.browser.page.keyboard.press("Enter")# 5. 等待并获取响应        response = await self._wait_for_response()return response

3. 具体的平台适配

每个平台都有针对性的实现，以DeepSeek为例：

# webot/deepseek/bot.pyclassDeepSeekBot(WebBot):"""DeepSeek机器人实现"""defrequires_login(self) -> bool:"""DeepSeek必须登录后才能使用"""returnTrueasyncdeflogin(self) -> bool:"""DeepSeek登录实现 - 手动登录方式"""        logger.debug(f"\n=== DeepSeek登录 ({self.config.name}) ===")# 如果已经在聊天页面，无需登录        current_url = self.browser.page.urlif current_url.startswith(self.config.chat_url):            self.is_logged_in = TruereturnTrue# 导航到登录页面await self.browser.goto(self.config.login_url)await self.browser.page.wait_for_load_state("networkidle")        print("请在浏览器中手动完成登录...")        print("登录完成后按回车键继续...")        input()  # 等待用户手动操作# 验证登录是否成功ifawait self._verify_login_complete():            self.is_logged_in = True            print("DeepSeek登录成功")returnTruereturnFalse

核心功能实现

浏览器驱动封装

drivers/browser.py提供了统一的浏览器操作接口：

classBrowserDriver:"""浏览器驱动封装，隐藏Playwright细节"""asyncdefstart(self, bot_name: str, save_login_state: bool = True):"""启动浏览器，支持状态保存和恢复"""try:# 1. 启动Playwright            self._playwright = await async_playwright().start()# 2. 启动Chromium，添加反检测参数            launch_args = ["--disable-blink-features=AutomationControlled","--start-maximized","--disable-popup-blocking","--disable-notifications",            ]            self._browser = await self._playwright.chromium.launch(                headless=self.headless,                args=launch_args,            )# 3. 加载或创建浏览器上下文（支持状态恢复）            self._context = await self._load_or_create_context(                bot_name, save_login_state            )# 4. 创建页面并添加反检测脚本            self._page = await self._context.new_page()            init_script = self.config.browser.init_scriptif init_script:await self._page.add_init_script(init_script)except Exception as e:await self._cleanup_on_error()  # 优雅的资源清理raise BrowserError(f"启动浏览器失败: {e}")asyncdefupload_files(self, selector: str, files: List[str]) -> bool:"""统一的上传文件接口"""# 验证文件存在for file_path in files:            path = Path(file_path)ifnot path.exists():raise FileError(f"文件不存在: {file_path}")# 使用Playwright的文件上传API        file_input = self.page.locator(selector)await file_input.wait_for(state="attached", timeout=5000)await file_input.set_input_files(files)# 等待上传完成await self.page.wait_for_timeout(1000)returnTrue

多文件处理

批量文件上传功能：

# webot/base/web_bot.pyasyncdef_upload_files(self, files: List[str]) -> List[str]:"""上传文件列表到聊天界面"""    upload_selector = self.config.selectors.get("file_upload")ifnot upload_selector:        logger.debug("未配置文件上传选择器，跳过文件上传")return []    uploaded_files = []for file_path in files:        path = Path(file_path)ifnot path.exists() ornot path.is_file():continuetry:# 使用浏览器驱动的统一上传接口await self.browser.upload_single_file(upload_selector, file_path)            uploaded_files.append(file_path)            logger.debug(f"已上传文件: {file_path}")# 文件间等待，避免过快触发限制await asyncio.sleep(2)except Exception as e:            logger.error(f"上传文件失败 {file_path}: {e}")return uploaded_files

按目录上传文件分析：

asyncdef_upload_directory(self, directory_path: str) -> List[str]:"""上传整个目录，自动生成结构说明文档"""    directory = Path(directory_path)ifnot directory.exists():raise FileNotFoundError(f"目录不存在: {directory_path}")# 1. 生成目录结构说明文件    structure_file = await self._create_directory_structure(directory)# 2. 获取目录下所有文件（智能过滤）    files = []for item in directory.rglob("*"):if item.is_file():# 过滤排除项if (item.suffix notin self.exclude_exts andnot any(excluded in item.parts for excluded in self.exclude_dirs)):                files.append(str(item))    print(f"发现 {len(files)} 个文件，准备上传...")# 3. 将结构文档作为第一个文件上传if structure_file:        files.insert(0, structure_file)# 4. 批量上传returnawait self._upload_files(files)

目录结构生成代码：

asyncdef_create_directory_structure(self, directory: Path) -> str:"""创建目录结构Markdown文档"""    output_file = "directory_structure.md"with open(output_file, "w", encoding="utf-8") as f:        f.write("=" * 60 + "\n")        f.write(f"目录结构: {directory}\n")        f.write(f"生成时间: {datetime.now():%Y-%m-%d %H:%M:%S}\n")        f.write("=" * 60 + "\n\n")# 使用递归生成树状结构defwrite_tree(folder: Path, prefix: str = "", is_last: bool = True):            items = sorted(folder.iterdir(), key=lambda x: (not x.is_dir(), x.name.lower()))for i, item in enumerate(items):                item_is_last = i == len(items) - 1                connector = "└── "if item_is_last else"├── "if item.is_dir():# 跳过排除目录if item.name in self.exclude_dirs:continue                    f.write(f"{prefix}{connector}{item.name}/\n")                    new_prefix = prefix + ("    "if item_is_last else"│   ")                    write_tree(item, new_prefix, item_is_last)else:# 跳过排除文件if item.suffix.lower() in self.exclude_exts:continue                    f.write(f"{prefix}{connector}{item.name}\n")        write_tree(directory)return output_file

平台差异化处理策略

不同AI平台的响应机制差异很大，需要针对性的等待策略：

DeepSeek的流式响应检测：

# webot/deepseek/bot.pyasyncdef_wait_for_response(self) -> str:"""等待DeepSeek的流式响应完成"""    max_stable_checks = 2# 最大稳定检查次数    last_new_text = ""    last_new_length = 0    stable_count = 0# 等待新响应开始    wait_start_time = time.time()while time.time() - wait_start_time < 20:        current_text = await self._get_new_response_only()if current_text and current_text != self.last_response_text:if len(current_text.strip()) > 0:                last_new_text = current_text                last_new_length = len(current_text)                response_started = Truebreakawait asyncio.sleep(0.5)# 监控响应进度whileTrue:        current_text = await self._get_new_response_only()        current_length = len(current_text)if current_length > last_new_length:# 有新内容生成            last_new_text = current_text            last_new_length = current_length            stable_count = 0elif current_length == last_new_length and current_length > 0:# 内容长度稳定            stable_count += 1if stable_count >= max_stable_checks:# 认为响应完成                self.last_response_text = current_textreturn current_textawait asyncio.sleep(1.0)

通义千问的停止按钮检测：

# webot/qianwen/bot.pyasyncdefcheck_has_stop_button(self, page):"""检查页面上是否有停止按钮"""try:        stop_selectors = ['div[class*="stop"]']for selector in stop_selectors:            elements = await page.locator(selector).all()if elements:for element in elements:ifawait element.is_visible():returnTruereturnFalseexcept Exception as e:returnFalseasyncdef_wait_for_response(self) -> str:"""等待千问响应完成"""# 1. 等待停止按钮出现（AI开始响应）    start_time = time.time()while time.time() - start_time < 30:        has_stop_button = await self.check_has_stop_button(self.browser.page)if has_stop_button:breakawait asyncio.sleep(0.5)# 2. 等待停止按钮消失（AI响应完成）    response_start_time = time.time()while time.time() - response_start_time < 30 * 60:        has_stop_button = await self.check_has_stop_button(self.browser.page)ifnot has_stop_button:# 停止按钮消失，验证是否有AI响应内容ifawait self.check_has_ai_response_content(self.browser.page, timeout=1000):breakawait asyncio.sleep(0.5)# 3. 获取Markdown格式内容returnawait self._read_copybutton_response()

配置管理

services/config_service.py实现了灵活的配置管理：

classConfigService:"""配置服务 - 支持YAML和JSON格式"""def__init__(self, config_dir: str = "configs") -> None:        self.config_dir = Path(config_dir)        self.config_dir.mkdir(exist_ok=True)defload(self, bot_name: str) -> BotConfig:"""加载机器人配置"""        config_file = self._find_config_file(bot_name)ifnot config_file:raise FileNotFoundError(f"配置文件不存在: {bot_name}")# 根据扩展名选择加载器if config_file.suffix.lower() in [".yaml", ".yml"]:            data = ConfigService.load_yaml(config_file)else:            data = ConfigService.load_json(config_file)# 转换浏览器配置        browser_data = data.get("browser", {})if browser_data:            browser_config = BrowserConfig(**browser_data)            data["browser"] = browser_config# 创建配置对象return BotConfig(**data)    @staticmethoddefload_yaml(file_path: Path) -> Dict[str, Any]:"""加载YAML文件，支持环境变量替换"""        content = file_path.read_text(encoding="utf-8")# 替换环境变量 ${VAR} 或 ${VAR:default}        pattern = r"\$\{([A-Za-z0-9_]+)(?::([^}]+))?\}"defreplace_env_var(match: re.Match) -> str:            var_name = match.group(1)            default_value = match.group(2)            env_value = os.getenv(var_name)if env_value isnotNone:return env_valueelif default_value isnotNone:return default_valueelse:return match.group(0)        content = re.sub(pattern, replace_env_var, content)return yaml.safe_load(content)

命令行接口实现

cli.py提供了丰富的命令行功能：

classCLI:"""命令行接口类，提供与AI机器人的交互功能"""def__init__(self, log_level: str = "INFO"):        self.factory = BotFactory()  # 创建工厂实例        self.available_bots = self.factory.list_all()  # 获取可用机器人        self.configs = self._load_configs()  # 加载配置        self.command_registry = self._init_commands()  # 初始化命令asyncdefrun_interactive(self):"""运行交互模式"""        self._print_banner()ifnot self.available_bots:            print("未找到可用的机器人配置")return# 显示机器人列表        print("可用机器人:")for i, bot_type in enumerate(self.available_bots, 1):            display_name = self.configs.get(bot_type, bot_type)            print(f"  {i}. {display_name} ({bot_type})")# 选择机器人并进入对话循环        bot_type = self._select_bot(self.available_bots)ifnot bot_type:return        bot = self.factory.create(bot_type)asyncwith bot as bot_instance:await bot_instance.ensure_ready()await self._conversation_loop(bot_instance, display_name)

支持的命令：

• list - 列出可用机器人
• ask <bot> <question> - 单次提问
• chat <bot> - 进入对话模式
• config show <bot> - 显示配置
• history list <bot> - 查看历史记录

应用举例

ask 功能

# 使用DeepSeek审查Python代码ai-webot ask deepseek "审查这段代码的性能问题" file:./main.py# 同时获取多个AI的意见ai-webot ask deepseek "这段代码有哪些改进空间" file:./utils.pyai-webot ask doubao "检查代码中的安全问题" file:./auth.pyai-webot ask qianwen "如何重构这个模块" file:./legacy_module.py

目录上传分析

# 分析整个项目并生成文档ai-webot ask deepseek "分析这个Django项目的结构" dir:./myproject# 输出结果包括：# 1. 自动生成的目录结构图# 2. 各文件功能分析# 3. 架构建议# 4. 依赖关系梳理

批量文件上传分析

# 批量上传多个文件进行分析ai-webot ask deepseek "分析这些数据文件" file:./data1.csv file:./data2.json file:./data3.txt# 或者使用通配符模式（需要Shell支持）for file in *.py; do  ai-webot ask deepseek "分析这个Python文件" file:"$file"done

开发总结

项目不足

本版本为首发版本，存在缺陷和瑕疵待完善。

技术难点与解决方案

难点1：不同平台的登录机制差异

• 解决方案：抽象出requires_login()和login()方法，各平台自行实现
• DeepSeek：需要手动登录，保持cookie状态
• 豆包/通义千问：支持游客模式，无需强制登录

难点2：响应等待策略不同

• DeepSeek：检测流式响应的文本变化
• 豆包：通过消息计数和元素可见性判断
• 通义千问：监测停止按钮状态变化

难点3：文件上传接口差异

• 统一方案：通过配置指定文件上传选择器
• 智能适配：通义千问区分文档和图片上传通道
• 批量处理：实现目录上传和结构生成

项目中的其他实践

1. 错误处理策略：

• 自定义异常体系（BrowserError、FileError）
• 资源清理确保无泄漏
• 友好的错误提示信息

2. 配置驱动设计：

• 所有平台特性通过配置文件定义
• 支持环境变量替换
• 便于添加新平台

3. 异步资源管理：

• 使用异步上下文管理器
• 确保浏览器资源正确释放
• 支持并发操作

项目价值

项目价值

1. 学习价值：

• 完整的Python项目开发流程
• Playwright实战经验
• 设计模式在实际项目中的应用
• 异步编程的最佳实践

2. 实用价值：

• 真实可用的多AI聚合工具
• 大幅提升文件处理效率
• 统一管理对话历史

3. 工程价值：

• 分层的架构设计
• 规范的代码组织
• 完整的错误处理

后续计划

1. 功能增强：

• 图形化交互模式
• 更智能的响应解析
• 更多的AI平台支持

2. 体验优化：

• 进度显示改进
• 错误恢复机制
• 配置向导工具

3. 技术升级：

• 性能优化
• 测试覆盖率提升
• 文档完善

后记

本项目从实际需求出发，通过合理的技术选型和架构设计，最终实现了一个实用、可扩展的AI聚合工具。开发过程中遇到的每个挑战都是学习的机会，每个解决方案都加深了对技术的理解。

开源这个项目的初衷，不仅是分享代码，更是分享一种"通过解决实际问题来学习技术"的方法论。希望这个项目能对正在学习Python自动化、Web爬虫、异步编程的开发者有所启发。

核心理念：最好的学习是创造有用的东西，最高的效率来自于解决自己的真实需求。

技术的学习永无止境，但每个解决实际问题的项目都是成长的重要里程碑。在编码中思考，在创造中学习，这才是程序员真正的成长之路。