起因一个很朴素的烦恼你是不是也这样想问一个技术问题习惯性地打开三四个 AI 网页——腾讯元宝问一遍DeepSeek 再问一遍来回切标签页对比谁答得更靠谱。烦是真烦但去接一堆 API Key、充值、管理密钥又觉得杀鸡用牛刀——毕竟这些网页版本身免费好用缺的只是并排看这一个功能。于是有了这个项目ai-web-hub。核心想法很简单——不接 API直接用 Playwright 驱动一个真实的 Chromium 浏览器替我在网页上手动操作元宝和 DeepSeek然后把两边的流式回答实时并排显示出来顺便包一层 OpenAI 兼容接口让 Cherry Studio、LobeChat 这些现成客户端也能直接把它当模型用。一句话总结用浏览器自动化把网页版 AI 伪装成 API。架构三层拆得很干净浏览器用户──HTTP──▶ FastAPI (WSL:8900) │ Orchestrator扇出到 N 个站点超时隔离事件合流 │ ┌─────────┴─────────┐ Adapter:元宝 Adapter:DeepSeek └─────────┬─────────┘ BrowserPool 每站点一个持久化浏览器 context技术栈选的是Python FastAPI playwright-python前端就是一个不需要构建链的单文件 HTML 原生 JS。够用就好。真正决定这个项目好不好维护的是下面这几个关键决策。六个关键决策,决定了这个项目能不能长期活下去决策一:登录态怎么处理自动化浏览器最头疼的就是登录——尤其是要扫码登录的中文站点。这里选择的方案是首次有头登录 持久化到本地目录:第一次手动跑一个有头浏览器,扫码登录一次,登录态(cookie、localStorage)就落在./profiles/yuanbao/这样的目录里。之后所有请求都复用这份登录态,全自动、无头运行。多亏了WSLg——WSL2 自带的 Wayland 显示服务,容器里跑的 Chromium 窗口可以直接显示在 Windows 桌面上,扫码登录不需要额外搭 VNC。这是这个方案能成立的关键前提之一。决策二:新增一个站点 新增一个文件所有站点相关的东西——选择器、发送逻辑、流式抽取、登录态判断——全部收在一个Adapter基类里,子类只需要给一份选择器字典。腾讯元宝的适配器完整代码就这么多:classYuanbaoAdapter(Adapter):nameyuanbaolabel腾讯元宝entry_urlos.getenv(HUB_YUANBAO_URL,https://yuanbao.tencent.com/chat/naQivTmsDa)# 联网搜索/思考阶段的状态文字,不能当成回答内容NOISE(正在搜索资料,正在搜索,搜索中,正在思考,思考中)SEL{# 元宝用的是富文本编辑器(Quill),不是 textareainput:[div.ql-editor[contenteditabletrue],div[contenteditabletrue][data-placeholder],div[class*chat-input] [contenteditabletrue],div[contenteditabletrue],textarea,],answer:[div.hyc-content-md,[class*hyc-content-md],],login_wall:[button.agent-dialogue__tool__login,[class*agent-dialogue__tool__login],[class*hyc-login],],send_button:[a[class*style__send-btn]:not([class*disabled]),[class*send-btn]:not([class*disabled]),],stop:[[class*style__stop],button[aria-label*停止],[class*chat-input-btn][class*stop],[class*icon-stop],],}注意每个 key 对应的不是一个选择器,而是一组候选选择器,按顺序取第一个命中的。基类的resolve()方法负责这件事:asyncdefresolve(self,page,key:str,*,timeout:float10.0,required:boolTrue,use_cache:boolTrue):返回 key 命中的 Locator。全部未命中时抛 SelectorMiss。 use_cacheFalse 时每次都按优先级重扫。answer 必须这样:精确候选往往比宽泛候选 晚出现(元宝的 hyc-content-md 要等真生成开始),缓存住宽泛候选就会一直读错节点。 cachedself._resolved.get(key)ifuse_cacheelseNoneifcached:try:ifawaitpage.locator(cached).count()0:returnpage.locator(cached)exceptException:pass# 缓存的候选失效了,往下走全量重试candidatesself.SEL.get(key,[])deadlinetime.monotonic()timeoutwhileTrue:forselincandidates:try:locpage.locator(sel)ifawaitloc.count()0:self._resolved[key]selreturnlocexceptException:continueiftime.monotonic()deadline:ifrequired:raiseSelectorMiss(key,candidates)returnNoneawaitasyncio.sleep(0.25)命中一次就缓存下来,下次直接用缓存的那个选择器,不用每次轮询都把候选列表重新试一遍——这在 250ms 一次的轮询频率下是个实打实的性能优化。browser.py和orchestrator.py完全不知道任何站点的选择器长什么样;反过来适配器也完全不碰并发、超时、SSE 这些脏活。以后想接一个新站点,大概率就是照抄yuanbao.py改一份SEL字典,不用动核心逻辑。决策三:流式回答怎么抓出来网页版 AI 没有现成的流式 API 给你订阅,只能靠盯屏幕。这里用的是轮询 前缀 diff:每 250ms 读一次回答区域的innerText,和上一次的内容做前缀比对,新增的部分就是这次的增量(delta)。如果站点半路重排了 DOM(比如 Markdown 渲染刷新导致新文本不再是旧文本的前缀),就退化成发一次全文覆盖(replace),前端整列刷新。这个方案比注入MutationObserver更抗折腾——虚拟 DOM 框架的重渲染经常会让 Observer 的事件变得不可靠。决策四:回答结束了没这件事,三重保险判断 AI 是不是说完了,单靠一个信号不够稳,这里叠了三层:信号说明主信号停止生成按钮消失 / 出现复制·重新生成操作栏兜底回答文本连续 1.5 秒无变化(应对站点改版导致按钮选择器失效)硬顶120 秒强制结束,返回已收到的部分文本三重信号和前缀 diff 增量抽取,实际都在同一个生成器函数里完成——这是整个项目最核心的一段逻辑:asyncdefstream_answer(self,page,index:int,*,timeout:float120.0,poll:float0.25,silence:float1.5,tail_grace:float0.2,)-AsyncIterator[StreamEvent]:starttime.monotonic()emittedlast_changestart saw_generatingFalsewhileTrue:textawaitself._read_answer(page,index)nowtime.monotonic()iftextandtext!emitted:iftext.startswith(emitted):yield(DELTA,text[len(emitted):])else:yield(REPLACE,text)emittedtext last_changenow generatingawaitself.is_generating(page)ifgenerating:saw_generatingTrueelifsaw_generatingand(emittedornow-startsilence):break# 主信号:曾在生成,现在停了# 主信号可用时把静默阈值放宽,避免模型中途思考停顿被误判为结束effective_silencesilence*4ifsaw_generatingelsesilenceifemittedandnow-last_changeeffective_silence:breakifnow-starttimeout:raiseasyncio.TimeoutError(f{self.name}超过{timeout}s 仍未结束)awaitasyncio.sleep(poll)# 收尾补抓,防止最后一个 chunk 漏掉awaitasyncio.sleep(tail_grace)finalawaitself._read_answer(page,index)iffinalandfinal!emitted:yield(DELTA,final[len(emitted):])iffinal.startswith(emitted)else(REPLACE,final)有几处细节是真实踩坑之后才补上的:saw_generating这个状态位记录是否真的观察到过生成中,避免一开始按钮还没渲染出来就被误判成已经结束主信号命中后,静默阈值会临时放宽到 4 倍(silence * 4),因为模型思考卡顿时文本会有短暂停滞,不能一停就判定说完了硬超时抛的是asyncio.TimeoutError,但抛出之前该 yield 的增量早就 yield 出去了——调用方拿到的是残缺but有内容的回答,而不是一个空异常再往下一层,_read_answer里还藏着一个容易被忽略的坑——联网搜索/思考阶段站点会往回答区塞正在搜索资料这类占位文字,如果把它当成正文,1.5 秒不变就会被静默兜底误判成说完了:asyncdef_read_answer(self,page,index:int)-str:locawaitself.resolve(page,answer,timeout0.2,requiredFalse,use_cacheFalse)iflocisNone:returnnodeloc.nth(index)try:ifawaitnode.count()0:returntextawaitnode.inner_text(timeout1500)exceptException:return# 状态占位文字不算回答:当成 才不会触发静默兜底,从而继续等真内容returniftext.strip()inself.NOISEelsetext这就是前面元宝适配器里NOISE (正在搜索资料, 正在思考, ...)那行的用途——一行配置,解决一个真实观察到的误判。决策五:并发模型——默认隔离,追问时才复用默认场景是扇出对比:一个问题同时甩给多个站点,每个站点开一个新标签页、新对话,互不干扰,方便并发。但如果想针对某个站点连续追问怎么办?加一个可选的session_id参数,复用同一个标签页做多轮对话。默认路径保持干净,特殊需求用一个参数覆盖——这是个很值得借鉴的 API 设计思路:不要为了支持 20% 的场景,把 80% 的默认路径搞复杂。决策六:选择器要挨骂,那就让它挨得体面点前端选择器是这类项目最大的长期维护成本——网站一改版,选择器就可能失效。应对方式很朴素但很管用:所有选择器集中放在每个适配器文件顶部的SEL字典里,改版了知道去哪改优先用data-testid、aria-label、role这类语义属性,不依赖会随构建变化的 class hash专门写了个scripts/probe.py,有头打开站点、发一句测试问题、把每条选择器的命中情况逐条打印出来——改版后 5 分钟内就能定位到底坏在哪一条Playwright 小课堂:内容到底是怎么抓出来的前面提到很多用 Playwright 做什么,这里花点篇幅讲讲Playwright 是怎么做到的——这几个原理理解了,遇到类似的自动化需求都能照搬。1. Locator 不是找到的元素,而是一个查找配方Playwright 的page.locator(selector)返回的不是某个具体的 DOM 节点,而是一份惰性的查找说明——每次你调用.count()、.click()、.inner_text(),它才真的去页面里查一次。这意味着:locpage.locator(div.hyc-content-md)awaitloc.count()# 这次查一次awaitloc.first.click()# 这次又查一次,拿到的可能是全新的 DOM 节点这个设计对付 SPA 特别合适——React/Vue 这类框架经常整段地销毁重建 DOM,如果像老式 Selenium 那样先拿到一个元素句柄再操作,句柄很容易变成悬空引用(StaleElementReferenceException)。Locator 每次都重新查,天然没有这个问题。项目里resolve()方法能做选择器候选回退 缓存命中的那一条,靠的就是这个特性。2. 为什么用inner_text(),不用text_content()或innerHTMLPlaywright 抓文本有三种典型方式,行为完全不同:方法拿到什么项目里为什么选/不选inner_text()渲染后用户实际看到的文字,经过 CSS 处理(display:none的内容不算,换行按渲染排版来)✅ 选它——最贴近人眼看到的回答内容text_content()DOM 树里的原始文本,包括被display:none隐藏的部分会把隐藏的骨架屏文字、tooltip 文案也读进来,脏evaluate(el el.innerHTML)原始 HTML 标签能读到最全信息,但得自己写 HTML→文本的转换,复杂度全转嫁到自己身上inner_text()底层其实是走 CDP(Chrome DevTools Protocol)在浏览器进程里真正计算一次布局,所以拿到的是所见即所得的文本——这也是为什么可以直接拿它当用户会读到的回答来做前缀 diff,不用自己处理换行、隐藏元素这些脏活。3. 为什么是轮询 diff,不是监听网络请求或MutationObserver这是这个项目里最容易被问到为什么不这样做的一处设计。摆在面前其实有三个选项:监听page.on(response),直接嗅探站点自己的流式接口——技术上可行,元宝、DeepSeek 内部大概率也是 SSE 或分块传输。但这条路的代价是:得反向工程每个站点私有的响应格式(通常还会变、可能加密混淆),站点一次内部重构就可能让解析全部失效,而且完全绑死在某个特定接口版本上。注入MutationObserver监听 DOM 变化——原理上更实时,但这些站点大多用虚拟 DOM 框架,一次 re-render 可能把整个回答节点连子树一起换掉,而不是增量 patch,Observer 收到的会是大量噪声事件,还得自己再做一次 diff——等于没省事。轮询inner_text() 应用层前缀 diff(项目选择的方案)——完全不关心页面内部用了什么框架、什么协议,只认最终渲染出来的文字。哪怕站点把前端框架从 React 换成 Vue,只要选择器还认得那个节点,这套逻辑照样能跑。选轮询本质上是用实时性换鲁棒性——250ms 的轮询间隔人眼完全感知不到延迟,但换来的是不必绑定任何站点的私有实现细节。这跟整个项目选择器改版了只改 SEL 字典的哲学是一致的:永远只依赖对方最外层、最稳定的那一层契约。4.launch_persistent_context:让浏览器记住登录状态的关键 API普通的browser.new_context()每次都是一个全新的、用完即弃的浏览器身份。项目里用的是另一个 API:ctxawaitself._pw.chromium.launch_persistent_context(user_data_dirstr(profile),# 例如 ./profiles/yuanbao/headlessself.s.headless,argsCHROME_ARGS,ignore_default_args[--enable-automation],viewport{width:1440,height:900},localezh-CN,timezone_idAsia/Shanghai,)user_data_dir就是真实 Chrome 用来存 Cookie、localStorage、IndexedDB、缓存的那个用户配置文件夹——跟你日常用的 Chrome 换个头像/账号切换配置文件是同一套机制。这意味着扫码登录一次之后,登录态是以整个浏览器身份的形式被保留下来的,而不是只存一个 token,所以比手动提取storageState再注入更接近真实用户的浏览器指纹,也更抗网站的风控检测。代价是:同一个user_data_dir同一时刻只能被一个 Chromium 进程独占——这也是为什么有头登录脚本运行前,必须先让服务侧的浏览器池放手(discard())。5. CDP 层的两个隐身小动作_STEALTH_JSObject.defineProperty(navigator, webdriver, {get: () undefined});...awaitctx.add_init_script(_STEALTH_JS)add_init_script底层调用的是 CDP 的Page.addScriptToEvaluateOnNewDocument——它保证这段 JS 会在页面自己的任何脚本运行之前注入执行。Playwright 默认会把navigator.webdriver设为true,这是最经典的自动化指纹,很多网站的风控第一步就检查它;这行代码把它重新伪装成undefined,伪装成一个正常人开的浏览器。配合ignore_default_args[--enable-automation](去掉 Chrome 顶部那条Chrome 正受到自动测试软件的控制提示条对应的启动参数)和--disable-blink-featuresAutomationControlled,三者一起把最基础的几层自动化指纹擦掉。这不是黑科技,只是把 Playwright 默认暴露的痕迹关掉而已——真要对付专业风控,项目里也留了后手(D4):patchright这个 API 完全兼容的反检测 fork,需要时改一行 import 就能换。6.page.route:在请求真正发出前拦截它asyncdef_route_blocker(route)-None:ifroute.request.resource_typeinBLOCKED_TYPES:# image / font / mediaawaitroute.abort()else:awaitroute.continue_()awaitpage.route(**/*,_route_blocker)这背后是 CDP 的Fetch.enable 请求拦截机制:注册后,页面发出的每一个网络请求都会先经过这个回调,你可以选择abort()(直接掐死,浏览器认为请求失败)、continue_()(放行)或者fulfill()(自己伪造一个响应)。项目里只用了最简单的abort/continue_,按resource_type过滤掉图片、字体、媒体请求——反正只是要读文字回答,这些资源加载了纯属浪费内存和带宽,在 WSL 只有 7GB 内存的环境下,这一行代码的省钱效果很直接。注意登录用的 tab 特意不开这个拦截,因为验证码本身就是图片,拦了就扫不了码。别忽略的工程细节单站点故障不传染每个站点跑在自己的asyncio.Task里,错误、超时都在自己的try/except里被圈住,转换成一个error/timeout事件塞进合流队列,不会让其他站点的任务受到牵连:try:...asyncforkind,textinadapter.stream_answer(page,index,...):charslen(text)ifkindREPLACEelsecharslen(text)emit(typekind,valuetext)okTrueemit(typedone,elapsedround(time.monotonic()-t0,2),charschars)exceptasyncio.TimeoutError:emit(typetimeout,partial_charschars,elapsedround(time.monotonic()-t0,2))exceptNotLoggedIn:emit(typeerror,codeNOT_LOGGED_IN,hintfdocker compose run --rm login --site{site})exceptSelectorMissasexc:emit(typeerror,codeSELECTOR_MISS,keyexc.key,hintf站点可能改版,跑 scripts/probe.py --site{site}定位)exceptExceptionasexc:emit(typeerror,codeINTERNAL,messagef{type(exc).__name__}:{exc})值得一提的是错误信息本身就是可操作的——NOT_LOGGED_IN直接把重登录命令带出来,SELECTOR_MISS直接告诉你该跑哪个排查脚本。前端拿到这些事件,不用额外查文档就知道下一步该干什么。浏览器崩了,自己重建Chromium 在资源紧张时偶尔会整个崩掉(Target closed之类的报错)。BrowserPool.new_page抓到这类错误后,会先把坏掉的 context 丢弃,重建一次再重试:asyncdefnew_page(self,site:str,*,block_resources:bool|NoneNone)-Page:blockself.s.block_resourcesifblock_resourcesisNoneelseblock_resourcesforattemptin(1,2):try:ctxawaitself.context(site)pageawaitctx.new_page()ifblock:awaitpage.route(**/*,_route_blocker)returnpageexceptExceptionasexc:ifattempt1and_is_closed_error(exc):log.warning(%s 的浏览器已崩溃重建后重试%s,site,exc)awaitself.discard(site)continueraiseraiseRuntimeError(unreachable)只重试一次——如果重建之后还是崩,大概率是环境问题(比如内存不够),再重试下去只是浪费时间,不如让错误暴露出来。两个容易被忽视但决定成败的小事shm_size: 2gb:容器默认/dev/shm只有 64MB,不设置这个 Chromium 会随机崩溃——这是真实踩过的坑,不是理论风险。启动参数里同时加了--disable-dev-shm-usage做双保险。测试重心放在可控的部分:真实网站不可控,所以核心测试是用本地的假聊天页(fixtures/fake_chat.html)模拟结束判定和前缀 diff 逻辑,而不是对着真实站点的 DOM 断言——那种测试注定会频繁失败且没有信息量。留一个scripts/smoke.py做人工验收的冒烟测试就够了。把网页封装成API 给本地其他服务调用前面讲的都是内部怎么实现的服务跑起来之后实际暴露给外部调用的接口专门伪装成 OpenAI 协议、给 Cherry Studio / LobeChat 这类现成客户端用的兼容接口。GET /v1/models和POST /v1/chat/completions—— OpenAI 兼容层这一组接口的意义是零改造接入现成客户端Cherry Studio、LobeChat 只要把 API 地址填成http://localhost:8900/v1就能把yuanbao、deepseek当成两个普通的模型来选。curlhttp://localhost:8900/v1/chat/completions\-HContent-Type: application/json\-d{ model: deepseek, messages: [{role: user, content: 11等于几}], stream: false }最有意思的是怎么把 OpenAI 的多轮消息数组翻译成这个项目的session_id概念——取messages里除最后一条外的全部消息序列化后取哈希当作session_iddef_session_from_prefix(messages:list[ChatMessage])-str|None:取除最后一条外的全部消息做哈希第二轮带完整历史时就能命中上一轮留下的 tab。prefixmessages[:-1]ifnotprefix:returnNonerawjson.dumps([m.model_dump()forminprefix],ensure_asciiFalse,sort_keysTrue)returnoai-hashlib.sha1(raw.encode(utf-8)).hexdigest()[:16]OpenAI 协议里客户端每轮都要把完整历史带回来而这个项目底层维护的其实是一个还活着的浏览器标签页。只要客户端带的历史前缀跟上一轮完全一致哈希就相同就能命中同一个session_id从而复用同一个标签页也就是网页里的同一个对话——只把最后一条新消息发进去不用把历史重新回放一遍给网页版 AI。首轮只有一条 user 消息时前缀为空哈希返回None直接走开新对话的路径。stream: true时走的是 SSE但要把内部的delta/replace/error事件翻译成 OpenAI 的chat.completion.chunk格式asyncdef_oai_stream(cid,created,site,prompt,session_id):...asyncforevinorch.ask_stream(prompt,[site],session_id):kindev.get(type)ifkinddelta:fullev[value]elifkindreplace:# OpenAI 协议无法撤回已发内容只能把超出已发长度的部分补上fullev[value]elifkinderror:yieldchunk({content:f\n\n[{ev.get(code)}]{detail}})continueelse:continueiflen(full)sent:yieldchunk({content:full[sent:]})sentlen(full)yieldchunk({},finishstop)yielddata: [DONE]\n\n这里有个协议层面天然的错位内部的replace事件表示站点重排了 DOM请用全文覆盖但 OpenAI 的流式协议是纯追加型的没有撤回已发内容这个概念。所以只能退而求其次——只把新全文里超出已发送长度的那部分补发出去如果重排后文本反而变短了就只能先忍着等finishstop兜底。这是两套协议模型不匹配时一个诚实的妥协而不是完美方案。不支持真 token 计数毕竟是从网页抓的文本没有模型侧的 tokenizerusage字段用字符数凑了个粗略估计能让客户端的用量统计不报错仅此而已。这个模块又是怎么调用 AI 网页的这是最容易被误解的一点——这里没有任何真正意义上的 API 调用。元宝和 DeepSeek 都没有对外开放的官方 API或者说即便有也不是这个项目想接的东西这个模块能做的一切都是假装自己是一个人在真实浏览器里点鼠标、敲键盘。一次请求从进来到拿到回答实际经过的是这样一条链路POST /api/ask │ ▼ Orchestrator._run_site(site, prompt) # app/orchestrator.py │ 1. broker.acquire(site) ──▶ 从 BrowserPool 要一个 Page │ 2. adapter.is_logged_in(page) 检查登录态 │ 3. adapter.answer_count(page) 先数一遍已有回答锁定本轮是第几条 │ 4. adapter.send(page, prompt) ───────────┐ │ 5. adapter.stream_answer(page, index) ◀────┘ ▼ YuanbaoAdapterapp/adapters/yuanbao.py base.py │ send(): 定位输入框 → box.fill(prompt) → keyboard.press(Enter) │ stream_answer(): 每 250ms page.locator(SEL[answer]).inner_text() │ 做前缀 diffyield 出增量 ▼ 真实的 ChromiumBrowserPool 里那个常驻的 persistent context │ 这是一个跟你自己电脑上的 Chrome 没有本质区别的浏览器进程 │ 只是启动参数带了 --headless且用的是扫码登录后攒下的用户目录 ▼ https://yuanbao.tencent.com ← 从这个网站的视角看进来的就是一个正常用户也就是说adapter.send()本质上就是这三行box(awaitself.resolve(page,input,timeout15.0)).firstawaitbox.click()awaitbox.fill(prompt)awaitpage.keyboard.press(Enter)跟你自己坐在电脑前打开元宝网页、点一下输入框、打字、按回车是完全相同的操作只是执行者从你的手换成了 Playwright 的自动化指令。读回答同理不是订阅了某个数据接口而是每 250ms 用inner_text()去读一次页面上那块回答区域渲染出来的文字——跟你自己盯着屏幕、隔一会儿瞟一眼答案打到哪儿了是同一件事只是这个模块做得比人眼更勤快、更规律。所以严格来说调用 AI 平台这个说法本身就不准确——这个项目根本没有调用任何 AI 平台的 API它调用的是浏览器操作的是网页。这也解释了为什么前面选择器、结束判定、防风控这些内容要占掉大半篇幅一旦选择自动化网页而不是接 API所有原本由官方 API 帮你兜底的稳定性版本兼容、结构化返回、限流提示都得自己在浏览器这一层重新造一遍。写在最后这个项目没有什么牛逼的技术,用的都是 Playwright、FastAPI 这些常规工具。真正值得琢磨的,是每一个不做什么和留一道兜底的选择:明确排除鉴权、上传、历史记录,把精力聚焦在能不能稳定跑起来这一件事上用一个接口把站点细节和核心逻辑彻底切开,让加站点变成体力活而不是脑力活给容易碎的地方(选择器、结束判定、DOM 抓取)都设计了兜底,而不是假设一切总能按预期工作如果你也受够了在标签页之间来回切换对比 AI 回答,这套用浏览器自动化伪装成 API的思路,或许值得一试。留个作业试试把Kimi 也接进来。需要源码的可以私聊我