1. 项目概述当自然语言指令遇上浏览器自动化最近在折腾一个挺有意思的东西我把它叫做“Qclaw智能体”。简单来说这玩意儿能让你在微信或者飞书里像跟同事聊天一样发句话比如“帮我查一下今天北京的天气截图发我”它就能在后台悄无声息地打开浏览器完成搜索、截图然后把结果通过聊天窗口回传给你。这听起来是不是有点像科幻电影里的场景其实这就是将自然语言处理NLP与浏览器自动化Browser Automation深度结合的一次实践。这个项目的核心价值在于它极大地降低了浏览器自动化操作的技术门槛。传统上无论是用Selenium、Puppeteer还是Playwright你都需要写代码来定义每一步操作打开哪个网址、点击哪个按钮、输入什么文字。而现在你只需要用最自然的方式说出你的需求。这对于那些需要频繁进行网页数据查询、信息填报、监控但又不具备编程能力的业务人员或者希望将一些固定、重复的网页操作流程“对话化”的开发者来说是一个效率利器。它把复杂的脚本编写变成了简单的对话交互。2. 核心架构与工作原理拆解要理解Qclaw智能体如何工作我们可以把它拆解成三个核心层交互层、大脑层和执行层。这三层协同工作将一句人话翻译成浏览器能执行的一系列动作。2.1 交互层多平台消息接入交互层负责与用户对话的“前台”。微信和飞书是两种非常典型的场景。微信代表了个人/轻量级办公场景而飞书则代表了企业级协同场景。实现上我们需要用到它们的开放平台接口。对于微信通常采用企业微信的“自建应用”或“群机器人”接口。你需要在企业微信后台创建一个应用获取到CorpID、Secret和AgentId。当用户在企微群里机器人或者私聊应用时微信服务器会将消息事件推送到你预设的接收URL通常是一个公网可访问的HTTP服务。你的服务端需要验证这个请求验证msg_signature等参数然后解析出用户ID、消息内容等信息。对于飞书逻辑类似通过创建“自定义机器人”来获取webhook地址或者创建“企业自建应用”来使用事件订阅。飞书的事件推送机制同样需要验证验证header中的x-lark-signature解析出event消息。注意消息接收服务必须部署在公网服务器并处理好安全验证否则无法正常接收消息。同时要妥善保管Secret等凭证避免泄露。2.2 大脑层意图理解与任务规划这是整个系统的“智能”所在。用户发来的“帮我查一下今天北京的天气截图发我”是一句非结构化的自然语言。大脑层需要做两件事意图识别和任务分解。意图识别通常借助大语言模型LLM的能力。我们可以将用户的原始指令连同一些预设的“技能”描述一起提交给LLM的API例如OpenAI的GPT系列、国内的一些大模型API。通过精心设计的提示词Prompt让模型判断用户的意图属于哪个预设类别并提取出关键参数。一个简单的Prompt示例可能是你是一个任务解析助手。请根据用户指令判断其意图并提取关键信息。 可用技能 1. 网页搜索截图技能名search_and_screenshot需要参数query搜索关键词。 2. 访问网页并提取文本技能名fetch_page_text需要参数url网页地址。 3. 登录网站并执行操作技能名login_and_action需要参数site网站标识action操作描述。 用户指令“帮我查一下今天北京的天气截图发我” 请以JSON格式输出包含字段intent技能名params参数字典。模型可能会返回{intent: search_and_screenshot, params: {query: 北京今天天气}}。任务分解则更进一步。对于复杂指令如“先去知乎搜一下AI编程的最新趋势把前三篇文章的标题和链接整理成表格发给我然后去GitHub看看今天trending的Python项目”单一的意图识别就不够了。这需要LLM进行多步任务规划生成一个有序的动作列表Action List。每个动作对应一个可执行的最小单元比如“打开浏览器访问知乎”、“在搜索框输入关键词‘AI编程 趋势’”、“提取搜索结果前三条的标题和链接”、“将数据格式化为Markdown表格”等。2.3 执行层无头浏览器的精准操控一旦大脑层规划好了具体的动作序列执行层就要负责将其落到实处。这里的主角是无头浏览器Headless Browser。我选择Playwright作为自动化工具因为它对现代Web技术如单页应用SPA支持更好API设计也更现代。执行层需要一个“驱动程序”来接收大脑层下发的动作指令并将其翻译成Playwright的API调用。例如动作“在搜索框输入关键词”可能对应着page.locator(input[nameq]).fill(keyword)。这个驱动程序需要维护浏览器实例Browser Context、页面Page的状态处理页面加载、元素等待、异常捕获等细节。一个关键的设计点是会话管理。当用户在微信里发起一个对话系统需要为这个对话创建一个独立的浏览器会话并在后续的交互中比如用户说“换上海试试”能够找到并复用这个会话直到任务结束或超时。这通常通过为每个用户或每个聊天线程分配一个唯一的session_id来实现并将该session_id与一个Playwright的BrowserContext关联起来。3. 核心模块实现与关键技术点理解了架构我们来看看几个核心模块的具体实现和那些容易踩坑的细节。3.1 消息接收与安全验证以飞书为例创建一个接收消息的HTTP服务使用FastAPI框架示意from fastapi import FastAPI, Request, HTTPException import hashlib import hmac import base64 import json app FastAPI() VERIFICATION_TOKEN your_verification_token # 飞书应用后台的Verification Token app.post(/webhook/lark) async def lark_webhook(request: Request): # 1. 验证签名 timestamp request.headers.get(X-Lark-Request-Timestamp) nonce request.headers.get(X-Lark-Request-Nonce) signature request.headers.get(X-Lark-Signature) body_bytes await request.body() # 拼接签名字符串 basestring f{timestamp}\n{nonce}\n{body_bytes.decode()}\n # 计算签名 hash_obj hmac.new(VERIFICATION_TOKEN.encode(), basestring.encode(), hashlib.sha256) computed_signature base64.b64encode(hash_obj.digest()).decode() if not hmac.compare_digest(computed_signature, signature): raise HTTPException(status_code403, detailInvalid signature) # 2. 解析事件 event_data json.loads(body_bytes) # 处理飞书URL验证挑战首次配置时 if event_data.get(type) url_verification: return {challenge: event_data.get(challenge)} # 3. 提取用户消息 event event_data.get(event) if event and event.get(type) message: user_id event.get(sender, {}).get(user_id) message_content json.loads(event.get(message, {}).get(content, {})).get(text) # 将 user_id, message_content 放入任务队列异步处理 # await task_queue.put((user_id, message_content)) return {msg: ok} return {msg: ignore}实操心得签名验证这一步千万不能省这是防止恶意请求伪造消息的第一道防线。另外飞书和微信的验证逻辑和字段名不同需要仔细阅读官方文档。消息处理建议采用异步队列如Redis RQ或Celery避免HTTP请求超时因为后续的LLM调用和浏览器操作可能很耗时。3.2 基于LLM的智能指令解析这是项目的“大脑”核心。我们利用LLM的上下文理解能力来解析指令。以下是一个更健壮的解析函数示例import openai import json import backoff # 一个更详细的技能描述系统 SKILL_DESCRIPTIONS 你是一个高级自动化助手。请根据用户指令选择最匹配的技能并提取所有必要参数。 技能列表 1. 技能名web_search_screenshot 描述使用搜索引擎进行搜索并对结果页面进行截图。 必需参数search_query (字符串搜索关键词) 可选参数search_engine (字符串默认baidu可选google, bing) 2. 技能名get_page_content 描述访问指定URL并提取页面的主要文本内容。 必需参数url (字符串有效的网页地址) 3. 技能名monitor_price 描述监控某个电商商品页面的价格变化。 必需参数product_url (字符串商品页URL), target_price (数字目标价格) ...更多技能 PROMPT_TEMPLATE f {SKILL_DESCRIPTIONS} 请严格按以下JSON格式输出不要有任何其他解释 {{ intent: 技能名, params: {{参数1: 值1, 参数2: 值2}}, confidence: 0.9, // 你对这个解析结果的置信度0-1之间 need_clarify: false, // 是否需要向用户澄清问题如参数缺失或模糊 clarify_question: // 如果需要澄清这里填写要问用户的问题 }} 用户指令{user_instruction} backoff.on_exception(backoff.expo, openai.error.RateLimitError, max_tries5) def parse_user_instruction(instruction: str) - dict: 解析用户指令返回结构化任务 try: response openai.ChatCompletion.create( modelgpt-3.5-turbo, # 或 gpt-4 messages[ {role: system, content: 你是一个精准的任务解析器。}, {role: user, content: PROMPT_TEMPLATE.format(user_instructioninstruction)} ], temperature0.1, # 低温度保证输出稳定性 max_tokens500 ) result_text response.choices[0].message.content.strip() # 处理可能出现的代码块标记 if result_text.startswith(json): result_text result_text[7:-3] if result_text.endswith() else result_text[7:] elif result_text.startswith(): result_text result_text[3:-3] if result_text.endswith() else result_text[3:] parsed_result json.loads(result_text) return parsed_result except json.JSONDecodeError as e: # 如果LLM返回的不是合法JSON可能是指令过于模糊 return { intent: unknown, params: {}, confidence: 0.0, need_clarify: True, clarify_question: f我无法准确理解您的指令‘{instruction}’能否请您说得更具体一些例如您想搜索什么或者操作哪个网站 } except Exception as e: # 处理其他异常如网络错误 raise注意事项LLM的调用有成本和延迟。务必设置合理的超时和重试机制。temperature参数设置为较低值如0.1-0.3可以使输出更确定、更符合格式要求。一定要做好异常处理当LLM返回非JSON或无法理解时要有降级策略比如返回一个要求用户澄清的响应。3.3 Playwright自动化执行引擎执行引擎是实干家。它接收结构化任务如{intent: web_search_screenshot, params: {search_query: 北京天气}}并执行对应的操作函数。from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError import asyncio from pathlib import Path class BrowserAutomationEngine: def __init__(self, session_id: str): self.session_id session_id self.context None self.page None self.screenshot_dir Path(f./screenshots/{session_id}) self.screenshot_dir.mkdir(parentsTrue, exist_okTrue) def start(self): 启动浏览器上下文 self.playwright sync_playwright().start() # 使用 headedFalse 无头模式在服务器上运行。调试时可设为True。 self.browser self.playwright.chromium.launch(headlessTrue, args[--disable-blink-featuresAutomationControlled]) # 每个会话一个独立的Context实现隔离 self.context self.browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ... ) self.page self.context.new_page() def execute_task(self, task: dict) - dict: 执行具体任务 intent task.get(intent) params task.get(params, {}) if intent web_search_screenshot: return self._web_search_and_screenshot(params) elif intent get_page_content: return self._fetch_page_content(params) # ... 其他技能分支 else: return {success: False, error: f未知的指令类型: {intent}} def _web_search_and_screenshot(self, params: dict) - dict: 执行搜索并截图 query params.get(search_query, ) engine params.get(search_engine, baidu) search_urls { baidu: fhttps://www.baidu.com/s?wd{query}, google: fhttps://www.google.com/search?q{query}, bing: fhttps://www.bing.com/search?q{query} } url search_urls.get(engine) if not url: return {success: False, error: f不支持的搜索引擎: {engine}} try: # 导航到搜索页面 self.page.goto(url, wait_untilnetworkidle) # 等待网络空闲 # 等待搜索结果主体加载 self.page.wait_for_selector(#content_left, timeout10000) # 百度结果区域选择器 # 截图 screenshot_path self.screenshot_dir / fsearch_{int(time.time())}.png self.page.screenshot(pathscreenshot_path, full_pageTrue) # 截取整个页面 # 可以顺便提取一些文本结果 search_results [] result_elements self.page.query_selector_all(h3 a) # 简单示例实际选择器需调整 for elem in result_elements[:5]: # 取前5个结果 title elem.text_content() link elem.get_attribute(href) if title and link: search_results.append({title: title.strip(), link: link}) return { success: True, message: f已完成对‘{query}’的搜索, data: { screenshot: str(screenshot_path.absolute()), top_results: search_results } } except PlaywrightTimeoutError: return {success: False, error: 页面加载或元素等待超时} except Exception as e: return {success: False, error: f自动化执行失败: {str(e)}} def _fetch_page_content(self, params: dict) - dict: 获取页面文本内容 # 实现类似访问URL使用page.text_content()或定位特定元素提取文本 pass def close(self): 清理资源 if self.page: self.page.close() if self.context: self.context.close() if self.browser: self.browser.close() if self.playwright: self.playwright.stop()踩坑实录1.反爬虫机制很多网站会检测Playwright/Puppeteer的自动化特征。通过args: [‘--disable-blink-featuresAutomationControlled’]和设置合理的user_agent可以缓解一部分但更复杂的网站可能需要使用更高级的隐身技术或代理IP。2.元素选择器不稳定依赖CSS选择器定位元素但网站前端可能随时改动。尽量使用># celery_app.py from celery import Celery from your_llm_parser import parse_user_instruction from your_browser_engine import BrowserAutomationEngine from your_message_sender import send_reply_to_user celery_app Celery(qclaw_tasks, brokerredis://localhost:6379/0, backendredis://localhost:6379/0) celery_app.task(bindTrue, max_retries3) def process_user_message(self, user_id: str, platform: str, message: str, session_id: str None): 处理用户消息的Celery任务 try: # 1. 解析指令 parsed_task parse_user_instruction(message) if parsed_task.get(need_clarify): # 需要澄清直接回复用户 send_reply_to_user(platform, user_id, parsed_task[clarify_question]) return # 2. 获取或创建浏览器会话 # 这里需要实现一个会话管理器将session_id映射到BrowserAutomationEngine实例 # 例如使用一个全局字典或Redis存储活跃会话 engine get_or_create_engine(session_id or user_id) # 3. 执行任务 result engine.execute_task(parsed_task) # 4. 格式化并发送回复 if result[success]: reply_msg f任务完成{result[message]} if result.get(data, {}).get(screenshot): # 上传截图到图床或临时存储并获取可访问的URL image_url upload_image(result[data][screenshot]) reply_msg f\n截图[查看]({image_url}) if result.get(data, {}).get(top_results): # 将搜索结果格式化为文本 results_text \n.join([f{i1}. {r[title]} for i, r in enumerate(result[data][top_results][:3])]) reply_msg f\n前三结果\n{results_text} else: reply_msg f任务执行失败{result[error]} send_reply_to_user(platform, user_id, reply_msg) # 5. 清理长时间未活动的会话 cleanup_inactive_sessions() except Exception as exc: # 任务失败重试 raise self.retry(excexc, countdown60)4.2 会话状态管理与资源回收会话管理是保证多用户并发和资源高效利用的关键。我们需要一个SessionManager来管理所有活跃的浏览器会话。import threading import time from datetime import datetime, timedelta class SessionManager: def __init__(self, session_timeout_minutes30): self._sessions {} # session_id - {engine: BrowserAutomationEngine, last_active: timestamp} self._lock threading.Lock() self.timeout timedelta(minutessession_timeout_minutes) def get_engine(self, session_id: str) - BrowserAutomationEngine: with self._lock: now datetime.now() # 清理过期会话 expired [sid for sid, data in self._sessions.items() if now - data[last_active] self.timeout] for sid in expired: data[engine].close() del self._sessions[sid] # 获取或创建新会话 if session_id not in self._sessions: engine BrowserAutomationEngine(session_id) engine.start() self._sessions[session_id] {engine: engine, last_active: now} else: self._sessions[session_id][last_active] now return self._sessions[session_id][engine] def cleanup_all(self): with self._lock: for sid, data in self._sessions.items(): data[engine].close() self._sessions.clear() # 全局单例管理器 session_manager SessionManager()实操心得会话超时时间需要根据实际场景调整。太短会导致用户多轮对话中断太长会占用过多服务器资源。可以考虑实现一个LRU最近最少使用淘汰机制而不是简单的定时清理。另外在Docker或Kubernetes部署时要注意浏览器实例的内存消耗较大需要为容器分配足够的内存建议至少1GB以上。4.3 部署与运维要点将整个系统部署到生产环境需要考虑以下几个要点依赖安装Playwright需要安装浏览器二进制文件。在Dockerfile中可以使用官方镜像或运行playwright install chromium命令。FROM python:3.10-slim RUN apt-get update apt-get install -y wget gnupg libnss3 libatk-bridge2.0-0 libxcomposite1 libxrandr2 libgbm1 libasound2 WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt RUN playwright install chromium --with-deps COPY . . CMD [gunicorn, app:app, -k, uvicorn.workers.UvicornWorker, --bind, 0.0.0.0:8000]横向扩展任务处理WorkerCelery Worker可以水平扩展但BrowserAutomationEngine实例本身是有状态的每个会话一个浏览器实例。一种模式是将会话状态集中存储如RedisWorker无状态每次处理任务时根据session_id从中心存储中恢复或创建浏览器上下文。但这会带来序列化/反序列化的开销。更常见的做法是使用“粘性会话”让同一用户的请求路由到同一个Worker实例。监控与日志必须建立完善的监控。记录关键指标消息接收量、LLM调用延迟与成功率、浏览器任务执行时长与成功率、各会话资源占用。日志要详细记录每个任务的输入、输出和错误信息便于排查问题。可以使用Prometheus Grafana进行指标可视化。成本控制LLM API调用和可能用到的云服务是主要成本。需要对指令解析和任务规划进行优化比如缓存常见指令的解析结果、设置单用户调用频率限制、对非关键任务使用成本更低的模型如GPT-3.5-Turbo而非GPT-4。5. 典型应用场景与扩展思路这个框架的潜力远不止于查天气和搜网页。结合不同的“技能”定义它可以化身无数个实用的自动化助手。5.1 企业内部效率工具场景HR助手新员工在群里说“帮我开通Jira、Confluence和GitLab账号”智能体自动访问内部IT系统填写表单完成审批流程触发并将工单号返回。数据日报机器人每天上午在群里说“生成昨天的销售数据报表”机器人自动登录CRM后台导出数据用Python进行简单分析生成图表和摘要发送到群内。IT运维助手“查看一下官网的当前访问状态”机器人自动访问几个关键健康检查页面和监控仪表盘截图并判断服务是否正常。5.2 个人生活与信息管理场景比价购物助手“帮我看看iPhone 15在京东和天猫的价格”自动打开两个电商网站搜索商品提取价格和促销信息整理后回复。内容聚合助手“今天科技圈有什么热点”自动访问几个预定的科技媒体、博客和Hacker News抓取头条标题和链接总结后发回。自动化签到助手对接定时任务Cron Job每天定点在指定网站完成签到任务并在签到失败时通过微信通知你。5.3 扩展方向让智能体更“智能”视觉理解能力CV当前主要依赖HTML结构操作。加入计算机视觉能力后智能体可以处理验证码识别通过集成OCR或打码平台、读取图片中的信息、甚至进行基于屏幕图像的更模糊的操作比如“点击那个蓝色的按钮”。记忆与上下文让智能体记住对话历史。例如用户先说“查北京天气”然后说“那上海呢”智能体能理解“那”指的是天气并自动将地点参数换成“上海”。这需要维护更复杂的对话状态。技能市场与插件化设计一个插件系统让开发者可以轻松贡献新的“技能”Skill。每个技能就是一个Python类实现标准的接口execute(params)。系统动态加载用户可以通过自然语言描述来发现和调用新技能。自动化流程编排从单条指令扩展到多步骤工作流。用户可以用自然语言描述一个完整的流程“每周一早上先去A网站下载数据报表然后登录B系统上传报表最后把成功结果发邮件给张三和李四。” 系统将其解析成一个可调度的工作流Workflow并按时执行。6. 常见问题与故障排查手册在实际开发和运行中你会遇到各种各样的问题。这里记录了一些典型问题及其解决方法。问题现象可能原因排查步骤与解决方案收不到微信/飞书消息1. 网络问题公网无法访问你的服务。2. 安全验证失败。3. 应用配置错误Token、URL等。1. 使用ngrok或frp等工具将本地服务临时暴露到公网测试。2. 检查日志中的签名验证错误对比官方文档逐字核对签名算法。3. 在平台后台检查接收消息的服务器地址配置是否正确是否有空格。LLM返回结果格式错误1. Prompt设计不佳导致模型不按JSON输出。2. 模型“幻觉”生成无关内容。3. 网络超时或API限制。1. 在Prompt中明确要求“只输出JSON”、“不要有任何解释”。使用json.loads()前尝试用正则提取{}之间的内容。2. 降低temperature参数值。在系统指令中强调“严格遵守格式”。3. 实现重试和退避机制检查API密钥的额度和频率限制。浏览器自动化失败元素找不到1. 页面未完全加载或动态加载。2. 选择器CSS/XPath写错了或已过时。3. 网站有反爬机制检测到自动化工具。1. 在操作前增加等待page.wait_for_selector(selector, statevisible, timeout10000)。使用wait_until: networkidle或domcontentloaded。2. 使用浏览器开发者工具重新检查元素优先使用有id或>任务执行超时1. 网络慢页面加载时间长。2. 某个步骤陷入死循环或等待条件永不满足。3. LLM响应慢。1. 为page.goto和wait_for_*函数设置合理的timeout参数如30秒。2. 为整个任务设置全局超时如Celery任务的soft_time_limit。在代码中加入超时判断和中断逻辑。3. 考虑将LLM调用与浏览器操作放在不同的异步任务中避免相互阻塞。服务器内存占用越来越高1. 浏览器实例Context/Page未正确关闭。2. 会话无限增长没有回收机制。1. 确保每个BrowserAutomationEngine在close()方法中按顺序关闭page, context, browser, playwright。2. 实现并严格执行会话超时清理机制如SessionManager。定期重启Worker进程也是一个简单粗暴但有效的方法。截图或结果无法发送回用户1. 媒体文件上传失败。2. 消息推送API调用失败权限、频率限制。3. 回复消息格式不符合平台要求。1. 先将截图保存到服务器本地或对象存储如S3、OSS生成一个可访问的临时URL再将URL发送给用户。2. 检查消息推送接口的返回错误码飞书和微信都有详细的错误码文档。注意access_token的获取和刷新。3. 飞书和微信对消息内容格式JSON结构要求严格务必按照官方示例构建消息体。开发这样一个智能体最大的挑战往往不在核心的自动化技术而在于稳定性和异常处理。网络是不稳定的网站是会改版的用户的指令是模糊多变的。因此系统的每一个环节都要有“防御性编程”的思想设想各种失败情况并给出优雅的降级方案比如明确地告诉用户“这个网站暂时无法访问”或“我没太听懂您能换个说法吗”这比直接抛出一个Python错误堆栈要友好得多。从我个人的经验来看从小而具体的场景开始打磨比如就先做好“搜索截图”这一件事跑通整个闭环然后再逐步添加新的技能和更复杂的逻辑是成功率最高的路径。当你看到第一句“今天天气怎么样”真的变成一张天气截图发回群里时那种感觉还是挺棒的。 基于LLM与Playwright的智能浏览器自动化：从自然语言到网页操作
1. 项目概述当自然语言指令遇上浏览器自动化最近在折腾一个挺有意思的东西我把它叫做“Qclaw智能体”。简单来说这玩意儿能让你在微信或者飞书里像跟同事聊天一样发句话比如“帮我查一下今天北京的天气截图发我”它就能在后台悄无声息地打开浏览器完成搜索、截图然后把结果通过聊天窗口回传给你。这听起来是不是有点像科幻电影里的场景其实这就是将自然语言处理NLP与浏览器自动化Browser Automation深度结合的一次实践。这个项目的核心价值在于它极大地降低了浏览器自动化操作的技术门槛。传统上无论是用Selenium、Puppeteer还是Playwright你都需要写代码来定义每一步操作打开哪个网址、点击哪个按钮、输入什么文字。而现在你只需要用最自然的方式说出你的需求。这对于那些需要频繁进行网页数据查询、信息填报、监控但又不具备编程能力的业务人员或者希望将一些固定、重复的网页操作流程“对话化”的开发者来说是一个效率利器。它把复杂的脚本编写变成了简单的对话交互。2. 核心架构与工作原理拆解要理解Qclaw智能体如何工作我们可以把它拆解成三个核心层交互层、大脑层和执行层。这三层协同工作将一句人话翻译成浏览器能执行的一系列动作。2.1 交互层多平台消息接入交互层负责与用户对话的“前台”。微信和飞书是两种非常典型的场景。微信代表了个人/轻量级办公场景而飞书则代表了企业级协同场景。实现上我们需要用到它们的开放平台接口。对于微信通常采用企业微信的“自建应用”或“群机器人”接口。你需要在企业微信后台创建一个应用获取到CorpID、Secret和AgentId。当用户在企微群里机器人或者私聊应用时微信服务器会将消息事件推送到你预设的接收URL通常是一个公网可访问的HTTP服务。你的服务端需要验证这个请求验证msg_signature等参数然后解析出用户ID、消息内容等信息。对于飞书逻辑类似通过创建“自定义机器人”来获取webhook地址或者创建“企业自建应用”来使用事件订阅。飞书的事件推送机制同样需要验证验证header中的x-lark-signature解析出event消息。注意消息接收服务必须部署在公网服务器并处理好安全验证否则无法正常接收消息。同时要妥善保管Secret等凭证避免泄露。2.2 大脑层意图理解与任务规划这是整个系统的“智能”所在。用户发来的“帮我查一下今天北京的天气截图发我”是一句非结构化的自然语言。大脑层需要做两件事意图识别和任务分解。意图识别通常借助大语言模型LLM的能力。我们可以将用户的原始指令连同一些预设的“技能”描述一起提交给LLM的API例如OpenAI的GPT系列、国内的一些大模型API。通过精心设计的提示词Prompt让模型判断用户的意图属于哪个预设类别并提取出关键参数。一个简单的Prompt示例可能是你是一个任务解析助手。请根据用户指令判断其意图并提取关键信息。 可用技能 1. 网页搜索截图技能名search_and_screenshot需要参数query搜索关键词。 2. 访问网页并提取文本技能名fetch_page_text需要参数url网页地址。 3. 登录网站并执行操作技能名login_and_action需要参数site网站标识action操作描述。 用户指令“帮我查一下今天北京的天气截图发我” 请以JSON格式输出包含字段intent技能名params参数字典。模型可能会返回{intent: search_and_screenshot, params: {query: 北京今天天气}}。任务分解则更进一步。对于复杂指令如“先去知乎搜一下AI编程的最新趋势把前三篇文章的标题和链接整理成表格发给我然后去GitHub看看今天trending的Python项目”单一的意图识别就不够了。这需要LLM进行多步任务规划生成一个有序的动作列表Action List。每个动作对应一个可执行的最小单元比如“打开浏览器访问知乎”、“在搜索框输入关键词‘AI编程 趋势’”、“提取搜索结果前三条的标题和链接”、“将数据格式化为Markdown表格”等。2.3 执行层无头浏览器的精准操控一旦大脑层规划好了具体的动作序列执行层就要负责将其落到实处。这里的主角是无头浏览器Headless Browser。我选择Playwright作为自动化工具因为它对现代Web技术如单页应用SPA支持更好API设计也更现代。执行层需要一个“驱动程序”来接收大脑层下发的动作指令并将其翻译成Playwright的API调用。例如动作“在搜索框输入关键词”可能对应着page.locator(input[nameq]).fill(keyword)。这个驱动程序需要维护浏览器实例Browser Context、页面Page的状态处理页面加载、元素等待、异常捕获等细节。一个关键的设计点是会话管理。当用户在微信里发起一个对话系统需要为这个对话创建一个独立的浏览器会话并在后续的交互中比如用户说“换上海试试”能够找到并复用这个会话直到任务结束或超时。这通常通过为每个用户或每个聊天线程分配一个唯一的session_id来实现并将该session_id与一个Playwright的BrowserContext关联起来。3. 核心模块实现与关键技术点理解了架构我们来看看几个核心模块的具体实现和那些容易踩坑的细节。3.1 消息接收与安全验证以飞书为例创建一个接收消息的HTTP服务使用FastAPI框架示意from fastapi import FastAPI, Request, HTTPException import hashlib import hmac import base64 import json app FastAPI() VERIFICATION_TOKEN your_verification_token # 飞书应用后台的Verification Token app.post(/webhook/lark) async def lark_webhook(request: Request): # 1. 验证签名 timestamp request.headers.get(X-Lark-Request-Timestamp) nonce request.headers.get(X-Lark-Request-Nonce) signature request.headers.get(X-Lark-Signature) body_bytes await request.body() # 拼接签名字符串 basestring f{timestamp}\n{nonce}\n{body_bytes.decode()}\n # 计算签名 hash_obj hmac.new(VERIFICATION_TOKEN.encode(), basestring.encode(), hashlib.sha256) computed_signature base64.b64encode(hash_obj.digest()).decode() if not hmac.compare_digest(computed_signature, signature): raise HTTPException(status_code403, detailInvalid signature) # 2. 解析事件 event_data json.loads(body_bytes) # 处理飞书URL验证挑战首次配置时 if event_data.get(type) url_verification: return {challenge: event_data.get(challenge)} # 3. 提取用户消息 event event_data.get(event) if event and event.get(type) message: user_id event.get(sender, {}).get(user_id) message_content json.loads(event.get(message, {}).get(content, {})).get(text) # 将 user_id, message_content 放入任务队列异步处理 # await task_queue.put((user_id, message_content)) return {msg: ok} return {msg: ignore}实操心得签名验证这一步千万不能省这是防止恶意请求伪造消息的第一道防线。另外飞书和微信的验证逻辑和字段名不同需要仔细阅读官方文档。消息处理建议采用异步队列如Redis RQ或Celery避免HTTP请求超时因为后续的LLM调用和浏览器操作可能很耗时。3.2 基于LLM的智能指令解析这是项目的“大脑”核心。我们利用LLM的上下文理解能力来解析指令。以下是一个更健壮的解析函数示例import openai import json import backoff # 一个更详细的技能描述系统 SKILL_DESCRIPTIONS 你是一个高级自动化助手。请根据用户指令选择最匹配的技能并提取所有必要参数。 技能列表 1. 技能名web_search_screenshot 描述使用搜索引擎进行搜索并对结果页面进行截图。 必需参数search_query (字符串搜索关键词) 可选参数search_engine (字符串默认baidu可选google, bing) 2. 技能名get_page_content 描述访问指定URL并提取页面的主要文本内容。 必需参数url (字符串有效的网页地址) 3. 技能名monitor_price 描述监控某个电商商品页面的价格变化。 必需参数product_url (字符串商品页URL), target_price (数字目标价格) ...更多技能 PROMPT_TEMPLATE f {SKILL_DESCRIPTIONS} 请严格按以下JSON格式输出不要有任何其他解释 {{ intent: 技能名, params: {{参数1: 值1, 参数2: 值2}}, confidence: 0.9, // 你对这个解析结果的置信度0-1之间 need_clarify: false, // 是否需要向用户澄清问题如参数缺失或模糊 clarify_question: // 如果需要澄清这里填写要问用户的问题 }} 用户指令{user_instruction} backoff.on_exception(backoff.expo, openai.error.RateLimitError, max_tries5) def parse_user_instruction(instruction: str) - dict: 解析用户指令返回结构化任务 try: response openai.ChatCompletion.create( modelgpt-3.5-turbo, # 或 gpt-4 messages[ {role: system, content: 你是一个精准的任务解析器。}, {role: user, content: PROMPT_TEMPLATE.format(user_instructioninstruction)} ], temperature0.1, # 低温度保证输出稳定性 max_tokens500 ) result_text response.choices[0].message.content.strip() # 处理可能出现的代码块标记 if result_text.startswith(json): result_text result_text[7:-3] if result_text.endswith() else result_text[7:] elif result_text.startswith(): result_text result_text[3:-3] if result_text.endswith() else result_text[3:] parsed_result json.loads(result_text) return parsed_result except json.JSONDecodeError as e: # 如果LLM返回的不是合法JSON可能是指令过于模糊 return { intent: unknown, params: {}, confidence: 0.0, need_clarify: True, clarify_question: f我无法准确理解您的指令‘{instruction}’能否请您说得更具体一些例如您想搜索什么或者操作哪个网站 } except Exception as e: # 处理其他异常如网络错误 raise注意事项LLM的调用有成本和延迟。务必设置合理的超时和重试机制。temperature参数设置为较低值如0.1-0.3可以使输出更确定、更符合格式要求。一定要做好异常处理当LLM返回非JSON或无法理解时要有降级策略比如返回一个要求用户澄清的响应。3.3 Playwright自动化执行引擎执行引擎是实干家。它接收结构化任务如{intent: web_search_screenshot, params: {search_query: 北京天气}}并执行对应的操作函数。from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError import asyncio from pathlib import Path class BrowserAutomationEngine: def __init__(self, session_id: str): self.session_id session_id self.context None self.page None self.screenshot_dir Path(f./screenshots/{session_id}) self.screenshot_dir.mkdir(parentsTrue, exist_okTrue) def start(self): 启动浏览器上下文 self.playwright sync_playwright().start() # 使用 headedFalse 无头模式在服务器上运行。调试时可设为True。 self.browser self.playwright.chromium.launch(headlessTrue, args[--disable-blink-featuresAutomationControlled]) # 每个会话一个独立的Context实现隔离 self.context self.browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ... ) self.page self.context.new_page() def execute_task(self, task: dict) - dict: 执行具体任务 intent task.get(intent) params task.get(params, {}) if intent web_search_screenshot: return self._web_search_and_screenshot(params) elif intent get_page_content: return self._fetch_page_content(params) # ... 其他技能分支 else: return {success: False, error: f未知的指令类型: {intent}} def _web_search_and_screenshot(self, params: dict) - dict: 执行搜索并截图 query params.get(search_query, ) engine params.get(search_engine, baidu) search_urls { baidu: fhttps://www.baidu.com/s?wd{query}, google: fhttps://www.google.com/search?q{query}, bing: fhttps://www.bing.com/search?q{query} } url search_urls.get(engine) if not url: return {success: False, error: f不支持的搜索引擎: {engine}} try: # 导航到搜索页面 self.page.goto(url, wait_untilnetworkidle) # 等待网络空闲 # 等待搜索结果主体加载 self.page.wait_for_selector(#content_left, timeout10000) # 百度结果区域选择器 # 截图 screenshot_path self.screenshot_dir / fsearch_{int(time.time())}.png self.page.screenshot(pathscreenshot_path, full_pageTrue) # 截取整个页面 # 可以顺便提取一些文本结果 search_results [] result_elements self.page.query_selector_all(h3 a) # 简单示例实际选择器需调整 for elem in result_elements[:5]: # 取前5个结果 title elem.text_content() link elem.get_attribute(href) if title and link: search_results.append({title: title.strip(), link: link}) return { success: True, message: f已完成对‘{query}’的搜索, data: { screenshot: str(screenshot_path.absolute()), top_results: search_results } } except PlaywrightTimeoutError: return {success: False, error: 页面加载或元素等待超时} except Exception as e: return {success: False, error: f自动化执行失败: {str(e)}} def _fetch_page_content(self, params: dict) - dict: 获取页面文本内容 # 实现类似访问URL使用page.text_content()或定位特定元素提取文本 pass def close(self): 清理资源 if self.page: self.page.close() if self.context: self.context.close() if self.browser: self.browser.close() if self.playwright: self.playwright.stop()踩坑实录1.反爬虫机制很多网站会检测Playwright/Puppeteer的自动化特征。通过args: [‘--disable-blink-featuresAutomationControlled’]和设置合理的user_agent可以缓解一部分但更复杂的网站可能需要使用更高级的隐身技术或代理IP。2.元素选择器不稳定依赖CSS选择器定位元素但网站前端可能随时改动。尽量使用># celery_app.py from celery import Celery from your_llm_parser import parse_user_instruction from your_browser_engine import BrowserAutomationEngine from your_message_sender import send_reply_to_user celery_app Celery(qclaw_tasks, brokerredis://localhost:6379/0, backendredis://localhost:6379/0) celery_app.task(bindTrue, max_retries3) def process_user_message(self, user_id: str, platform: str, message: str, session_id: str None): 处理用户消息的Celery任务 try: # 1. 解析指令 parsed_task parse_user_instruction(message) if parsed_task.get(need_clarify): # 需要澄清直接回复用户 send_reply_to_user(platform, user_id, parsed_task[clarify_question]) return # 2. 获取或创建浏览器会话 # 这里需要实现一个会话管理器将session_id映射到BrowserAutomationEngine实例 # 例如使用一个全局字典或Redis存储活跃会话 engine get_or_create_engine(session_id or user_id) # 3. 执行任务 result engine.execute_task(parsed_task) # 4. 格式化并发送回复 if result[success]: reply_msg f任务完成{result[message]} if result.get(data, {}).get(screenshot): # 上传截图到图床或临时存储并获取可访问的URL image_url upload_image(result[data][screenshot]) reply_msg f\n截图[查看]({image_url}) if result.get(data, {}).get(top_results): # 将搜索结果格式化为文本 results_text \n.join([f{i1}. {r[title]} for i, r in enumerate(result[data][top_results][:3])]) reply_msg f\n前三结果\n{results_text} else: reply_msg f任务执行失败{result[error]} send_reply_to_user(platform, user_id, reply_msg) # 5. 清理长时间未活动的会话 cleanup_inactive_sessions() except Exception as exc: # 任务失败重试 raise self.retry(excexc, countdown60)4.2 会话状态管理与资源回收会话管理是保证多用户并发和资源高效利用的关键。我们需要一个SessionManager来管理所有活跃的浏览器会话。import threading import time from datetime import datetime, timedelta class SessionManager: def __init__(self, session_timeout_minutes30): self._sessions {} # session_id - {engine: BrowserAutomationEngine, last_active: timestamp} self._lock threading.Lock() self.timeout timedelta(minutessession_timeout_minutes) def get_engine(self, session_id: str) - BrowserAutomationEngine: with self._lock: now datetime.now() # 清理过期会话 expired [sid for sid, data in self._sessions.items() if now - data[last_active] self.timeout] for sid in expired: data[engine].close() del self._sessions[sid] # 获取或创建新会话 if session_id not in self._sessions: engine BrowserAutomationEngine(session_id) engine.start() self._sessions[session_id] {engine: engine, last_active: now} else: self._sessions[session_id][last_active] now return self._sessions[session_id][engine] def cleanup_all(self): with self._lock: for sid, data in self._sessions.items(): data[engine].close() self._sessions.clear() # 全局单例管理器 session_manager SessionManager()实操心得会话超时时间需要根据实际场景调整。太短会导致用户多轮对话中断太长会占用过多服务器资源。可以考虑实现一个LRU最近最少使用淘汰机制而不是简单的定时清理。另外在Docker或Kubernetes部署时要注意浏览器实例的内存消耗较大需要为容器分配足够的内存建议至少1GB以上。4.3 部署与运维要点将整个系统部署到生产环境需要考虑以下几个要点依赖安装Playwright需要安装浏览器二进制文件。在Dockerfile中可以使用官方镜像或运行playwright install chromium命令。FROM python:3.10-slim RUN apt-get update apt-get install -y wget gnupg libnss3 libatk-bridge2.0-0 libxcomposite1 libxrandr2 libgbm1 libasound2 WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt RUN playwright install chromium --with-deps COPY . . CMD [gunicorn, app:app, -k, uvicorn.workers.UvicornWorker, --bind, 0.0.0.0:8000]横向扩展任务处理WorkerCelery Worker可以水平扩展但BrowserAutomationEngine实例本身是有状态的每个会话一个浏览器实例。一种模式是将会话状态集中存储如RedisWorker无状态每次处理任务时根据session_id从中心存储中恢复或创建浏览器上下文。但这会带来序列化/反序列化的开销。更常见的做法是使用“粘性会话”让同一用户的请求路由到同一个Worker实例。监控与日志必须建立完善的监控。记录关键指标消息接收量、LLM调用延迟与成功率、浏览器任务执行时长与成功率、各会话资源占用。日志要详细记录每个任务的输入、输出和错误信息便于排查问题。可以使用Prometheus Grafana进行指标可视化。成本控制LLM API调用和可能用到的云服务是主要成本。需要对指令解析和任务规划进行优化比如缓存常见指令的解析结果、设置单用户调用频率限制、对非关键任务使用成本更低的模型如GPT-3.5-Turbo而非GPT-4。5. 典型应用场景与扩展思路这个框架的潜力远不止于查天气和搜网页。结合不同的“技能”定义它可以化身无数个实用的自动化助手。5.1 企业内部效率工具场景HR助手新员工在群里说“帮我开通Jira、Confluence和GitLab账号”智能体自动访问内部IT系统填写表单完成审批流程触发并将工单号返回。数据日报机器人每天上午在群里说“生成昨天的销售数据报表”机器人自动登录CRM后台导出数据用Python进行简单分析生成图表和摘要发送到群内。IT运维助手“查看一下官网的当前访问状态”机器人自动访问几个关键健康检查页面和监控仪表盘截图并判断服务是否正常。5.2 个人生活与信息管理场景比价购物助手“帮我看看iPhone 15在京东和天猫的价格”自动打开两个电商网站搜索商品提取价格和促销信息整理后回复。内容聚合助手“今天科技圈有什么热点”自动访问几个预定的科技媒体、博客和Hacker News抓取头条标题和链接总结后发回。自动化签到助手对接定时任务Cron Job每天定点在指定网站完成签到任务并在签到失败时通过微信通知你。5.3 扩展方向让智能体更“智能”视觉理解能力CV当前主要依赖HTML结构操作。加入计算机视觉能力后智能体可以处理验证码识别通过集成OCR或打码平台、读取图片中的信息、甚至进行基于屏幕图像的更模糊的操作比如“点击那个蓝色的按钮”。记忆与上下文让智能体记住对话历史。例如用户先说“查北京天气”然后说“那上海呢”智能体能理解“那”指的是天气并自动将地点参数换成“上海”。这需要维护更复杂的对话状态。技能市场与插件化设计一个插件系统让开发者可以轻松贡献新的“技能”Skill。每个技能就是一个Python类实现标准的接口execute(params)。系统动态加载用户可以通过自然语言描述来发现和调用新技能。自动化流程编排从单条指令扩展到多步骤工作流。用户可以用自然语言描述一个完整的流程“每周一早上先去A网站下载数据报表然后登录B系统上传报表最后把成功结果发邮件给张三和李四。” 系统将其解析成一个可调度的工作流Workflow并按时执行。6. 常见问题与故障排查手册在实际开发和运行中你会遇到各种各样的问题。这里记录了一些典型问题及其解决方法。问题现象可能原因排查步骤与解决方案收不到微信/飞书消息1. 网络问题公网无法访问你的服务。2. 安全验证失败。3. 应用配置错误Token、URL等。1. 使用ngrok或frp等工具将本地服务临时暴露到公网测试。2. 检查日志中的签名验证错误对比官方文档逐字核对签名算法。3. 在平台后台检查接收消息的服务器地址配置是否正确是否有空格。LLM返回结果格式错误1. Prompt设计不佳导致模型不按JSON输出。2. 模型“幻觉”生成无关内容。3. 网络超时或API限制。1. 在Prompt中明确要求“只输出JSON”、“不要有任何解释”。使用json.loads()前尝试用正则提取{}之间的内容。2. 降低temperature参数值。在系统指令中强调“严格遵守格式”。3. 实现重试和退避机制检查API密钥的额度和频率限制。浏览器自动化失败元素找不到1. 页面未完全加载或动态加载。2. 选择器CSS/XPath写错了或已过时。3. 网站有反爬机制检测到自动化工具。1. 在操作前增加等待page.wait_for_selector(selector, statevisible, timeout10000)。使用wait_until: networkidle或domcontentloaded。2. 使用浏览器开发者工具重新检查元素优先使用有id或>任务执行超时1. 网络慢页面加载时间长。2. 某个步骤陷入死循环或等待条件永不满足。3. LLM响应慢。1. 为page.goto和wait_for_*函数设置合理的timeout参数如30秒。2. 为整个任务设置全局超时如Celery任务的soft_time_limit。在代码中加入超时判断和中断逻辑。3. 考虑将LLM调用与浏览器操作放在不同的异步任务中避免相互阻塞。服务器内存占用越来越高1. 浏览器实例Context/Page未正确关闭。2. 会话无限增长没有回收机制。1. 确保每个BrowserAutomationEngine在close()方法中按顺序关闭page, context, browser, playwright。2. 实现并严格执行会话超时清理机制如SessionManager。定期重启Worker进程也是一个简单粗暴但有效的方法。截图或结果无法发送回用户1. 媒体文件上传失败。2. 消息推送API调用失败权限、频率限制。3. 回复消息格式不符合平台要求。1. 先将截图保存到服务器本地或对象存储如S3、OSS生成一个可访问的临时URL再将URL发送给用户。2. 检查消息推送接口的返回错误码飞书和微信都有详细的错误码文档。注意access_token的获取和刷新。3. 飞书和微信对消息内容格式JSON结构要求严格务必按照官方示例构建消息体。开发这样一个智能体最大的挑战往往不在核心的自动化技术而在于稳定性和异常处理。网络是不稳定的网站是会改版的用户的指令是模糊多变的。因此系统的每一个环节都要有“防御性编程”的思想设想各种失败情况并给出优雅的降级方案比如明确地告诉用户“这个网站暂时无法访问”或“我没太听懂您能换个说法吗”这比直接抛出一个Python错误堆栈要友好得多。从我个人的经验来看从小而具体的场景开始打磨比如就先做好“搜索截图”这一件事跑通整个闭环然后再逐步添加新的技能和更复杂的逻辑是成功率最高的路径。当你看到第一句“今天天气怎么样”真的变成一张天气截图发回群里时那种感觉还是挺棒的。