手里这台华硕路由器在我这儿之前的正经工作就是管管Wi-Fi、挂挂硬盘。直到我把一个轻量级AI提示流编排器塞进去它才算是真正意义上参与了家里的“智能中枢”建设。这期开源系列第12篇就来复盘一下整个过程怎么在Merlin固件上把一个OpenAI兼容的AI入口做成局域网边缘网关让家里任何设备改一行base_url就能走同一条提示词流水线。整个过程不依赖Docker不依赖重型框架纯Python标准库就能跑。如果你手里正好有一台刷了Merlin的华硕路由器最近又在折腾本地大模型或云端API这篇东西大概率能帮你省下一块树莓派的钱。适合喜欢折腾固件、想玩边缘AI、又不想给家里再添一台常驻服务器的朋友。1. 思路拆解路由器为什么能干AI的活1.1 提示流编排器到底编排了什么先把这个概念掰开说。提示流编排器英文可以叫Prompt Flow Orchestrator它做的事情不是简单的请求转发而是把一个自然语言请求拆成一条可声明、可复用、可观测的处理流水线。每条流水线由若干个步骤组成比如先对用户输入做关键词分类再套一个系统提示词模板然后决定调用哪个模型后端最后对模型返回内容做长度截断或格式整理。整个过程就像工厂里的流水线每个工位干一件事顺序可以配置。为什么要做这一步因为当家里有多个设备需要调用AI能力时每个设备各自接模型API会非常乱。手环App要摘要、NAS要分类文件、智能音箱要回答天气、开发板要跑对话每个地方都写死一个API地址和模型参数改一次模型就要逐个设备去更新。把提示流编排器放在路由器上等于把“如何组织提示词、用哪个模型、做什么后处理”这些策略集中到了一个地方。设备端只需要知道一个地址、一个Key剩下的事情全部由网关处理。再加上“边缘网关”这个定位它的价值就更明显了。设备到路由器是局域网内的极低延迟路由器再根据配置决定请求是扔给家里那台跑着Ollama的机器还是交给云端的大模型服务。这样既保住了隐私又能兜底复杂任务。说白了路由器在这里扮演的是AI请求的“前台接待员”它不负责动脑子做大模型推理但它知道每位访客应该被带到哪个房间。1.2 华硕路由器Merlin的资源边界做这件事之前必须先搞清楚一个核心问题华硕路由器到底有多少余量能跑这套东西。以我手里这台RT-AX86U为例博通四核CPU、512MB内存这个配置在路由器里算中上游但和正经服务器比起来还是紧巴巴的。让路由器本地推理一个大模型肯定不现实这不是跑不跑得动的问题而是内存和散热都会被瞬间打爆。Merlin固件给这个路子提供了三样关键能力一是jffs持久化分区路由器重启后脚本和配置不丢二是Entware包管理器类似一个小型应用商店能装Python、curl、jq这些常用工具三是自定义启动脚本钩子比如services-start、post-mount这些允许我们在开机阶段执行自己的命令。这三样组合起来路由器就不再是一台只能配置Wi-Fi的设备而是一个有完整运行时的边缘Linux环境。但资源边界始终在那儿。我实测下来一个纯Python实现的编排器进程常驻内存大约60到80MB这在一个512MB内存的路由器上是可接受的但绝对不能放任它膨胀。所以整个项目的设计原则很明确轻量、克制、单进程、低依赖。路由器适合做“控制面”而不是“数据面”也就是说它负责认知和协调重活让背后的模型服务器去干。这个思维转变很关键别一上来就想什么都塞进去。1.3 方案选型为什么是Python标准库而不是Docker全家桶写这篇文章之前其实也纠结过技术栈。第一个念头是Docker毕竟现在很多跑在NAS上的服务都是容器化部署但Merlin固件对Docker的支持有很大的硬件限制。华硕官方有一些x86型号的路由器可以玩Docker可绝大多数博通ARM平台的路由器根本没有这个选项即使强行装上容器运行时本身就要吃一两百MB内存还没跑业务就先被系统杀进程了。所以直接在项目方案里把Docker排除掉这不是技术洁癖是硬件条件不允许。第二个念头是用Node.js或者Go写这个网关。Node在Entware里确实有但常驻内存同样不低而且npm生态在路由器上装依赖是个折磨人的过程。Go编译出来虽然是个静态二进制很优雅但交叉编译和调试流程在家庭折腾场景下还是有点繁琐。最后我选了Python标准库方案核心原因是Python自带http.server、sqlite3、json、urllib这些模块一个不需要pip install任何第三方包的项目在Entware这个包不齐全的环境里就是神。配置格式也定了JSON而不是YAML省掉PyYAML这个依赖换来的是配置文件不能写注释的代价完全值得。并发模型也做了取舍。没有上FastAPI、uvicorn这类异步框架就用标准库的ThreadingHTTPServer配合一个信号量限制并发数为4。家庭局域网场景同时来十几个请求已经了不得了线程模型完全够用而且内存占用远低于异步框架。做嵌入式环境的服务最重要的不是框架够不够潮而是能不能在256MB到1GB的内存里稳定存活。2. 环境准备给你的路由器装个应用商店2.1 固件与基础设置在动手写代码之前得先把路由器的底子打好。我的RT-AX86U刷的是Merlin 388.4.x分支不同型号的刷机流程不太一样这里不展开讲刷机步骤假设你已经有一台跑着Merlin的华硕路由器。刷好之后首先要做的两件事是开启SSH和JFFS分区。SSH开启路径在管理界面里的“系统→服务→SSH”建议端口别用默认的22换一个高位端口密码也用强度高一点的因为路由器长期在线任何暴露在外面的端口都会被脚本扫描。JFFS的开启路径在“系统→JFFS2”勾选Enable JFFS2第一次开启的时候可以顺便勾上Format JFFS2让它格式化出一块干净空间。JFFS是存在路由器内部flash上的一小块可写区域容量通常只有几十MB不适合放项目本体但放启动脚本和配置足够了。接下来是USB存储。我这边挂了一块128GB的U盘格式化成了ext4分区。提醒一句别用NTFS虽然Merlin能读NTFS但小文件读写性能和权限管理都不如ext4舒服。U盘插到路由器的USB口之后SSH进路由器执行df -h确认系统识别到了盘符。这一步很重要后续Entware装到哪个分区完全取决于这里你看到的是什么。2.2 部署Entware运行时Entware是Merlin生态里最核心的包管理器装了它路由器才真正有了类似Linux发行版的软件源。在SSH终端里运行固件自带的entware-setup.sh脚本它会扫描已插入的USB设备列出所有可用分区让你选择。选分区的时候要注意Entware会在这个分区下创建/opt目录结构也就是说这个分区实际上成了路由器的软件运行盘。安装完成之后检查一下/opt/bin/opkg是否存在。然后更新软件源并安装基本工具opkg update opkg install python3 curl jq这里我特意没有安装python3-pip原因前面说过了整个项目零第三方依赖。如果你后续想扩展别的功能再装pip也不迟。curl和jq是排查问题时非常趁手的工具一个用来发测试请求一个用来在终端里格式化JSON输出。装好之后/opt/bin/python3 --version应该能正常输出版本号。这个版本大概率是Python 3.8或3.9写代码的时候要避免用太新的语法特性。Entware装好之后一个常见问题是路由器重启后/opt目录是空的因为Entware是装在USB设备上的系统还没来得及挂载。Merlin固件一般会在开机阶段自动挂载USB并激活Entware但这个动作有延迟。所以后面写自启动脚本的时候一定要加sleep等待逻辑否则脚本跑得比挂载快就会找不到python3。2.3 项目目录与swap规划Entware就位后开始规划项目目录。我的结构很简单所有东西都放在/opt/ai-gateway/下面/opt/ai-gateway/ ├── main.py # 入口服务 ├── flow.py # 提示流引擎 ├── cache.py # SQLite缓存模块 ├── config.json # 声明式配置 ├── start.sh # 启动脚本 ├── watchdog.sh # 守护脚本 └── logs/ # 日志目录代码放在USB盘上而不是jffs里一方面是因为jffs空间太小另一方面是频繁读写flash会加速颗粒老化。U盘虽然也会磨损但坏了更换成本低得多而且AI网关这种服务读多写少压力不大。还有一个容易被忽略的环节是swap。华硕路由器内存虽然不小但Python解释器加上各种网络连接内存占用还是会慢慢爬升。强烈建议在USB盘上创建一个512MB的swap文件给系统一个缓冲余地。创建命令如下dd if/dev/zero of/opt/ai-gateway/swapfile bs1M count512 mkswap /opt/ai-gateway/swapfile swapon /opt/ai-gateway/swapfile开机自动挂载swap可以写到/jffs/scripts/post-mount这个钩子里在Entware挂载完成后执行swapon /opt/ai-gateway/swapfile。这个步骤能显著降低OOM杀进程的概率尤其是你同时还在路由器上跑着其他插件的时候。3. 提示流编排器核心设计3.1 一条请求的生命周期现在进入正题。理解这套系统的运行机制可以把一次完整的请求生命周期拆开看这是整个设计的骨架。第一步局域网内的某个设备向http://192.168.50.1:8787/v1/chat/completions发出一个POST请求请求头和OpenAI官方格式一致带着Authorization: Bearer开头的API Key。第二步服务端先校验API Key是否合法再校验来源IP是否在允许的网段内。这一步相当于门卫把陌生人挡在门外。第三步限流模块检查这个Key在当前时间窗口内是否已经刷了太多请求防的是某个应用死循环把云端API额度打爆。第四步从请求体里提取最后一条用户消息这是提示流处理的原料。第五步进入核心的提示流执行阶段。编排器读取config.json里的流程定义按顺序执行每个步骤。典型流程是先做规则路由根据关键词决定走哪个模型后端然后做模板渲染把用户消息嵌入到一个系统提示词里接着调用选定的后端模型接口拿到结果后做后处理比如截断长度、剥离多余换行。第六步如果启用了缓存系统会先查询缓存命中则直接返回结果不命中才调用后端。最后把模型返回的内容包装成OpenAI格式的响应写回给请求方。每一条请求都走这六步每一步都可以在配置里开关或调整。这种设计把“策略”和“实现”完全分离改行为不用动代码只改JSON这在家用路由器这种不方便登录调试的环境里是非常实用的特性。3.2 用JSON定义一条提示流配置是整个系统的灵魂。我把config.json写得尽量语义化让人一眼能看懂这条流水线的走向。贴一个实际在用的精简版{ server: { lan_ip: 192.168.50.1, port: 8787, api_keys: [edge-gw-2024], allowed_subnets: [192.168.50.0/24] }, rate_limit: { per_key_per_minute: 30 }, cache: { enable: true, ttl_seconds: 3600 }, backends: { home_ollama: { type: openai_compatible, base_url: http://192.168.50.10:11434/v1, model: qwen3:8b, timeout: 30 }, cloud_model: { type: openai_compatible, base_url: https://api.example.com/v1, model: gpt-4o-mini, timeout: 20 } }, flows: { default: { steps: [ { op: route, rules: [ { pattern: 天气|日历|提醒, backend: home_ollama } ], default_backend: cloud_model }, { op: prompt, template: 你是家庭AI助手。请用中文简洁回答控制在100字以内。用户问题{{ last_user_message }} }, { op: call_backend }, { op: trim, max_chars: 500 } ] } } }解释一下几个关键字段。backends定义了可选的模型后端目前统一用openai_compatible类型这意味着Ollama、vLLM、各类云端模型API只要是OpenAI格式都能接进来。timeout按后端单独设置本地模型给长一点云端服务短一点。flows.default.steps定义了默认提示流的四个步骤先根据正则规则选后端然后渲染模板接着调用后端最后修剪输出长度。这个配置有一个很实用的细节route里的pattern是正则表达式。如果用户输入包含“天气”或“日历”这类关键词请求会被路由到本地Ollama因为这类简单任务本地模型响应快、零成本。其他复杂任务走default_backend即云端模型。这样既省钱又分流是边缘网关的核心卖点之一。3.3 轻量网关的三张牌鉴权、限流、缓存一个网关如果只有转发功能那没有任何价值。这套系统在工程上之所以成立靠的是鉴权、限流、缓存这三张牌。鉴权做的是双层校验。第一层是API Key校验请求头里的Bearer token必须和config.json里的api_keys匹配。第二层是来源IP校验只有allowed_subnets里的网段才能访问服务。路由器上做IP校验有个天然优势它自己就是网关能明确知道请求是从哪个接口进来的。家里内网设备走LAN口访客网络走另一个网段可以在配置里只放行主网段让访客设备即使拿到Key也调不通。限流用的是内存滑动窗口。按API Key维度记录每分钟的请求时间戳超过阈值就返回429状态码。家庭场景默认每分钟30次已经非常宽裕但对异常程序来说这个限制能在云端账单爆炸之前踩刹车。实现的时候不需要Redis一个Python dict加一个deque就够用了毕竟这个网关只服务一个家庭局域网几十个并发就已经是极限场景。缓存是目前实际收益最大的一环。实现思路很直接把用户消息和流程名拼接后做SHA256哈希把哈希作为Key把模型返回结果和过期时间存进SQLite。TTL默认一小时。为什么这个环节收益大因为家庭成员对AI问的问题高度重复“今天有什么安排”这种话每天要说一遍第一遍花了3秒调用云端模型之后59分钟内的重复提问都是毫秒级返回。既省了云端API调用费又让设备端体验好了不少。4. 代码落地与部署实战4.1 核心代码入口服务项目核心代码分三个文件main.py负责HTTP入口flow.py负责提示流引擎cache.py负责缓存。main.py的核心结构非常精简标准库自带的http.server加ThreadingHTTPServer就能撑起整个服务。精简版代码大致长这样#!/usr/bin/env python3 import json import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer CONFIG json.load(open(/opt/ai-gateway/config.json, encodingutf-8)) SEMAPHORE threading.BoundedSemaphore(4) class GatewayHandler(BaseHTTPRequestHandler): def do_POST(self): if self.path ! /v1/chat/completions: self.send_error(404) return if not self._check_auth(): self.send_error(401) return length int(self.headers.get(Content-Length, 0)) body json.loads(self.rfile.read(length).decode(utf-8)) user_text body[messages][-1][content] with SEMAPHORE: result run_flow(default, user_text, CONFIG) payload json.dumps({ id: chatcmpl-edge, object: chat.completion, model: result[backend], choices: [{ index: 0, message: {role: assistant, content: result[text]}, finish_reason: stop }] }, ensure_asciiFalse).encode(utf-8) self.send_response(200) self.send_header(Content-Type, application/json; charsetutf-8) self.send_header(Content-Length, str(len(payload))) self.end_headers() self.wfile.write(payload) def _check_auth(self) - bool: token self.headers.get(Authorization, ) return token Bearer CONFIG[server][api_keys][0] if __name__ __main__: server ThreadingHTTPServer((CONFIG[server][lan_ip], CONFIG[server][port]), GatewayHandler) server.daemon_threads True server.serve_forever()重点说几个工程细节。信号量SEMAPHORE限制同一时间最多4个请求在跑这个数字是根据内存余量实测出来的。如果放开到几十个并发Python线程的内存开销加网络缓冲很容易把路由器的512MB内存吃穿。do_POST里先鉴权再解析body顺序不能反否则未授权请求也能触发JSON解析开销等于给了攻击者一个廉价的CPU消耗入口。响应体用的ensure_asciiFalse这个很重要否则中文会被转成\uXXXX的转义序列虽然客户端能正常解析但日志排查时看到全是一堆反斜杠头都要大。4.2 核心代码提示流引擎flow.py是整套系统的逻辑核心它负责把配置里的步骤一步步真正跑起来。核心函数run_flow的设计思路是遍历所有步骤用一个上下文ctx字典传递中间状态。贴关键片段def run_flow(flow_name: str, user_text: str, cfg: dict) - dict: steps cfg[flows][flow_name][steps] ctx {user_text: user_text} backend None for step in steps: op step[op] if op route: backend _route(step, user_text) ctx[backend] backend elif op prompt: ctx[prompt] _render(step[template], ctx) elif op call_backend: ctx[response] _call_backend(backend, ctx[prompt], cfg) elif op trim: ctx[response] ctx[response][: step.get(max_chars, 500)] return {text: ctx[response], backend: backend} def _route(step: dict, text: str) - str: for rule in step[rules]: if re.search(rule[pattern], text, re.I): return rule[backend] return step.get(default_backend, cloud_model) def _render(template: str, ctx: dict) - str: return template.replace({{ last_user_message }}, ctx[user_text])_route的逻辑很直白逐个规则做正则匹配第一个命中的后端返回。没有命中任何规则就落到默认后端。_render用的是最简单的字符串替换而不是Jinja2模板引擎这是故意这么做的路由器上没必要为一个模板功能引入几百KB的第三方库。_call_backend的实现用的是标准库的urllib.requestdef _call_backend(backend: str, prompt: str, cfg: dict) - str: conf cfg[backends][backend] url conf[base_url].rstrip(/) /chat/completions payload { model: conf[model], messages: [{role: user, content: prompt}] } req urllib.request.Request( url, datajson.dumps(payload).encode(utf-8), headers{Content-Type: application/json} ) with urllib.request.urlopen(req, timeoutconf.get(timeout, 30)) as resp: data json.loads(resp.read().decode(utf-8)) return data[choices][0][message][content]这个函数做的事情就是透传模型调用把渲染好的提示词发给后端然后从OpenAI格式的响应里取出文本返回。timeout从配置里读取本地Ollama给30秒云端模型给20秒超时时间必须单独控制否则一个慢后端会拖住整个网关的连接池。4.3 部署到路由器与自启动插件代码本地写完不是终点要让它成为路由器上的常驻服务还得做两步一是把文件传上去二是注册开机自启。上传我用的是SCP命令scp -r ai-gateway admin192.168.50.1:/opt/传完之后建议先在路由器上执行一遍语法检查cd /opt/ai-gateway python3 -m compileall .compileall能快速抓住缩进错误、语法错误这类低级问题比启动服务后在日志里翻报错高效得多。没问题之后手动跑一次start.sh看看能否正常拉起#!/bin/sh export PATH/opt/bin:/opt/sbin:/usr/sbin:/usr/bin:/sbin:/bin export PYTHONIOENCODINGutf-8 export LANGC.UTF-8 cd /opt/ai-gateway nohup python3 main.py logs/stdout.log 21 启动后立刻curl探一下端口确认服务已经起来了。开机自启我推荐用Entware的init脚本机制而不是直接写/jffs/scripts/services-start。原因很简单Entware有它自己一套服务管理流程在/opt/etc/init.d/目录下放一个S99aigw脚本如下#!/bin/sh PROCSmain.py ARGS/opt/ai-gateway/main.py DESCAI edge gateway这个脚本配合Entware的rc.common框架会在系统启动阶段自动被调用而且会在Entware挂载完成之后才执行避开了手动写services-start最头疼的时序问题。当然有些人更喜欢在jffs里加自己的启动钩子也不冲突只是要注意加sleep至少20秒给USB挂载和Entware启动留时间。还有一个防御手段是watchdog脚本。路由器上任何服务都有可能因为内存紧张被系统杀掉所以我另写了一个watchdog.sh每隔30秒检查一次进程是否存在不存在就重新拉起配合cru定时任务注册到系统cron里cru a aigw_watchdog */1 * * * * /opt/ai-gateway/watchdog.sh这样即使进程半夜被OOM杀掉最多一分钟内就能自动恢复不影响第二天早上的使用。4.4 局域网客户端接入实测部署完成后的验证环节我会分两个层面做测试。第一层是直接用curl发原始请求快速确认网关的鉴权、路由、调用链路是否正常curl -s http://192.168.50.1:8787/v1/chat/completions \ -H Authorization: Bearer edge-gw-2024 \ -H Content-Type: application/json \ -d {model:auto,messages:[{role:user,content:明天天气怎么样}]}如果配置的路由规则把“天气”归到本地Ollama这条请求会在一两秒内返回而且响应里能直接看到用的后端是home_ollama。第二层测试是用OpenAI的官方Python SDK接入这一层决定了家里其他设备接入是否方便。SDK只需要改base_url和api_key两个参数其他代码完全不用动from openai import OpenAI client OpenAI( base_urlhttp://192.168.50.1:8787/v1, api_keyedge-gw-2024 ) resp client.chat.completions.create( modelauto, messages[{role: user, content: 帮我把明天的待办整理成三点}] ) print(resp.choices[0].message.content)实测下来OpenAI SDK只需要这一个改动就能完整接入这意味着家里所有原本为云端API开发的脚本、Home Assistant的对话组件、甚至是某些开源聊天机器人的前端都能无缝切换到这个局域网网关。唯一要注意的是模型名填auto或者随便填网关不会真的把这个字段透传给后端后端选择完全由提示流的路由规则决定。5. 常见问题与排查实录5.1 服务起不来或开机丢失这是路由器自启服务最经典的问题。现象通常是刚部署完手动启动一切正常但重启路由器后服务就没了或者启动失败。排查方向很明确先看进程在不在再看日志有没有输出。第一步执行pgrep -f main.py如果没有输出说明进程压根没有拉起来。第二步检查/opt/bin/python3存不存在很多情况下重启后Entware还没挂载完成服务脚本已经跑了自然找不到解释器。第三种情况是启动脚本执行了但Python报错直接退出去logs/stdout.log里看有没有Traceback多半能定位到问题。最终的解决办法就是在自启动脚本里加足够的sleep等待或者改用Entware的init脚本机制来托管。5.2 路由器内存不足导致进程被杀如果你的路由器只有256MB内存这个坑几乎躲不掉。现象是进程运行一段时间后突然消失dmesg里能看到OOM killer的记录或者系统日志里出现“Out of memory”字样。我用free -m观察过Python进程加上内核缓存内存占用很容易攀升到接近上限。解决手段有三板斧。第一创建swap文件并确保开机自动启用给OOM killer一个缓冲。第二收紧服务端的并发信号量从4改成2虽然极限并发能力下降但内存增长会平缓很多。第三定期清理日志文件长时间运行的stdout.log会无脑膨胀在cron里加一条日志清理命令几行代码的事但能避免磁盘和内存双保险的压力。5.3 后端连接超时或返回异常网关本身没挂但设备端报错“connection error”或者“timeout”这时候要逐层排查。我的排查顺序是先越过网关直连后端确认模型服务本身活着再通过网关调一次看具体卡在哪一步。直连后端的命令很简单以本地Ollama为例在路由器上或者局域网内其他机器执行curl -s http://192.168.50.10:11434/v1/chat/completions \ -H Content-Type: application/json \ -d {model:qwen3:8b,messages:[{role:user,content:ping}]}如果直连返回正常说明问题出在网关和后端之间的超时配置。解决办法是调大config.json里对应后端的timeout值本地模型推理大上下文时确实需要更宽裕的时间。如果直连也不通检查后端服务的监听地址是否绑定在了正确的网卡上以及有没有防火墙拦截。云端API返回4xx错误的情况也遇到过。最常见的是模型名写错、上下文长度超限、或者某个必需参数缺失。因为网关是透传状态码的设备端看到什么错误直接按状态码去查原服务的文档就行。5.4 中文乱码与编码问题中文乱码的问题一半出在终端显示一半出在Python编码设置。如果你在SSH终端里手动启动服务然后发现终端打印的中文全是方块或者问号别急着改代码先检查终端软件自身的编码设置。我现在用的终端强制UTF-8编码专治路由器SSH会话里的中文显示问题。另一半的问题在服务本身。Python在写入日志、输出响应时默认编码可能不是UTF-8尤其是在Entware这种精简环境下。解决办法是在start.sh里加上两行环境变量export PYTHONIOENCODINGutf-8 export LANGC.UTF-8响应体的JSON序列化用ensure_asciiFalse这样返回的中文就是原始字符而不是转义序列。还有一个排查经验是日志里看到的乱码和实际API响应里的乱码没有必然联系客户端拿到什么数据要以实际抓包为准不要被SSH终端显示问题误导。最后分享一点实际体会这期项目前前后后折腾了两周踩过不少坑之后最大的一个体会是在资源受限的环境里做设计砍需求的能力比加功能的能力更重要。路由器上跑的服务必须把“配置驱动、单进程、零第三方依赖”当铁律来执行凡是有一条不满足都会在之后某个半夜变成让你爬起来救火的坑。也正因为有这些限制反而逼出了一个比服务器上跑的大体量方案更干净、更清晰的设计。目前这套网关在我家里已经稳定跑了一个多月下一步我打算用cru定时任务做一个早晨自动生成当日简报的流程再把它接进Home Assistant的对话组件里让路由器在家庭自动化里能再多管几件事。如果你有更好的想法也可以在这个思路上继续扩展。
华硕路由器变身边缘AI网关：Merlin固件上部署轻量级提示流编排器
手里这台华硕路由器在我这儿之前的正经工作就是管管Wi-Fi、挂挂硬盘。直到我把一个轻量级AI提示流编排器塞进去它才算是真正意义上参与了家里的“智能中枢”建设。这期开源系列第12篇就来复盘一下整个过程怎么在Merlin固件上把一个OpenAI兼容的AI入口做成局域网边缘网关让家里任何设备改一行base_url就能走同一条提示词流水线。整个过程不依赖Docker不依赖重型框架纯Python标准库就能跑。如果你手里正好有一台刷了Merlin的华硕路由器最近又在折腾本地大模型或云端API这篇东西大概率能帮你省下一块树莓派的钱。适合喜欢折腾固件、想玩边缘AI、又不想给家里再添一台常驻服务器的朋友。1. 思路拆解路由器为什么能干AI的活1.1 提示流编排器到底编排了什么先把这个概念掰开说。提示流编排器英文可以叫Prompt Flow Orchestrator它做的事情不是简单的请求转发而是把一个自然语言请求拆成一条可声明、可复用、可观测的处理流水线。每条流水线由若干个步骤组成比如先对用户输入做关键词分类再套一个系统提示词模板然后决定调用哪个模型后端最后对模型返回内容做长度截断或格式整理。整个过程就像工厂里的流水线每个工位干一件事顺序可以配置。为什么要做这一步因为当家里有多个设备需要调用AI能力时每个设备各自接模型API会非常乱。手环App要摘要、NAS要分类文件、智能音箱要回答天气、开发板要跑对话每个地方都写死一个API地址和模型参数改一次模型就要逐个设备去更新。把提示流编排器放在路由器上等于把“如何组织提示词、用哪个模型、做什么后处理”这些策略集中到了一个地方。设备端只需要知道一个地址、一个Key剩下的事情全部由网关处理。再加上“边缘网关”这个定位它的价值就更明显了。设备到路由器是局域网内的极低延迟路由器再根据配置决定请求是扔给家里那台跑着Ollama的机器还是交给云端的大模型服务。这样既保住了隐私又能兜底复杂任务。说白了路由器在这里扮演的是AI请求的“前台接待员”它不负责动脑子做大模型推理但它知道每位访客应该被带到哪个房间。1.2 华硕路由器Merlin的资源边界做这件事之前必须先搞清楚一个核心问题华硕路由器到底有多少余量能跑这套东西。以我手里这台RT-AX86U为例博通四核CPU、512MB内存这个配置在路由器里算中上游但和正经服务器比起来还是紧巴巴的。让路由器本地推理一个大模型肯定不现实这不是跑不跑得动的问题而是内存和散热都会被瞬间打爆。Merlin固件给这个路子提供了三样关键能力一是jffs持久化分区路由器重启后脚本和配置不丢二是Entware包管理器类似一个小型应用商店能装Python、curl、jq这些常用工具三是自定义启动脚本钩子比如services-start、post-mount这些允许我们在开机阶段执行自己的命令。这三样组合起来路由器就不再是一台只能配置Wi-Fi的设备而是一个有完整运行时的边缘Linux环境。但资源边界始终在那儿。我实测下来一个纯Python实现的编排器进程常驻内存大约60到80MB这在一个512MB内存的路由器上是可接受的但绝对不能放任它膨胀。所以整个项目的设计原则很明确轻量、克制、单进程、低依赖。路由器适合做“控制面”而不是“数据面”也就是说它负责认知和协调重活让背后的模型服务器去干。这个思维转变很关键别一上来就想什么都塞进去。1.3 方案选型为什么是Python标准库而不是Docker全家桶写这篇文章之前其实也纠结过技术栈。第一个念头是Docker毕竟现在很多跑在NAS上的服务都是容器化部署但Merlin固件对Docker的支持有很大的硬件限制。华硕官方有一些x86型号的路由器可以玩Docker可绝大多数博通ARM平台的路由器根本没有这个选项即使强行装上容器运行时本身就要吃一两百MB内存还没跑业务就先被系统杀进程了。所以直接在项目方案里把Docker排除掉这不是技术洁癖是硬件条件不允许。第二个念头是用Node.js或者Go写这个网关。Node在Entware里确实有但常驻内存同样不低而且npm生态在路由器上装依赖是个折磨人的过程。Go编译出来虽然是个静态二进制很优雅但交叉编译和调试流程在家庭折腾场景下还是有点繁琐。最后我选了Python标准库方案核心原因是Python自带http.server、sqlite3、json、urllib这些模块一个不需要pip install任何第三方包的项目在Entware这个包不齐全的环境里就是神。配置格式也定了JSON而不是YAML省掉PyYAML这个依赖换来的是配置文件不能写注释的代价完全值得。并发模型也做了取舍。没有上FastAPI、uvicorn这类异步框架就用标准库的ThreadingHTTPServer配合一个信号量限制并发数为4。家庭局域网场景同时来十几个请求已经了不得了线程模型完全够用而且内存占用远低于异步框架。做嵌入式环境的服务最重要的不是框架够不够潮而是能不能在256MB到1GB的内存里稳定存活。2. 环境准备给你的路由器装个应用商店2.1 固件与基础设置在动手写代码之前得先把路由器的底子打好。我的RT-AX86U刷的是Merlin 388.4.x分支不同型号的刷机流程不太一样这里不展开讲刷机步骤假设你已经有一台跑着Merlin的华硕路由器。刷好之后首先要做的两件事是开启SSH和JFFS分区。SSH开启路径在管理界面里的“系统→服务→SSH”建议端口别用默认的22换一个高位端口密码也用强度高一点的因为路由器长期在线任何暴露在外面的端口都会被脚本扫描。JFFS的开启路径在“系统→JFFS2”勾选Enable JFFS2第一次开启的时候可以顺便勾上Format JFFS2让它格式化出一块干净空间。JFFS是存在路由器内部flash上的一小块可写区域容量通常只有几十MB不适合放项目本体但放启动脚本和配置足够了。接下来是USB存储。我这边挂了一块128GB的U盘格式化成了ext4分区。提醒一句别用NTFS虽然Merlin能读NTFS但小文件读写性能和权限管理都不如ext4舒服。U盘插到路由器的USB口之后SSH进路由器执行df -h确认系统识别到了盘符。这一步很重要后续Entware装到哪个分区完全取决于这里你看到的是什么。2.2 部署Entware运行时Entware是Merlin生态里最核心的包管理器装了它路由器才真正有了类似Linux发行版的软件源。在SSH终端里运行固件自带的entware-setup.sh脚本它会扫描已插入的USB设备列出所有可用分区让你选择。选分区的时候要注意Entware会在这个分区下创建/opt目录结构也就是说这个分区实际上成了路由器的软件运行盘。安装完成之后检查一下/opt/bin/opkg是否存在。然后更新软件源并安装基本工具opkg update opkg install python3 curl jq这里我特意没有安装python3-pip原因前面说过了整个项目零第三方依赖。如果你后续想扩展别的功能再装pip也不迟。curl和jq是排查问题时非常趁手的工具一个用来发测试请求一个用来在终端里格式化JSON输出。装好之后/opt/bin/python3 --version应该能正常输出版本号。这个版本大概率是Python 3.8或3.9写代码的时候要避免用太新的语法特性。Entware装好之后一个常见问题是路由器重启后/opt目录是空的因为Entware是装在USB设备上的系统还没来得及挂载。Merlin固件一般会在开机阶段自动挂载USB并激活Entware但这个动作有延迟。所以后面写自启动脚本的时候一定要加sleep等待逻辑否则脚本跑得比挂载快就会找不到python3。2.3 项目目录与swap规划Entware就位后开始规划项目目录。我的结构很简单所有东西都放在/opt/ai-gateway/下面/opt/ai-gateway/ ├── main.py # 入口服务 ├── flow.py # 提示流引擎 ├── cache.py # SQLite缓存模块 ├── config.json # 声明式配置 ├── start.sh # 启动脚本 ├── watchdog.sh # 守护脚本 └── logs/ # 日志目录代码放在USB盘上而不是jffs里一方面是因为jffs空间太小另一方面是频繁读写flash会加速颗粒老化。U盘虽然也会磨损但坏了更换成本低得多而且AI网关这种服务读多写少压力不大。还有一个容易被忽略的环节是swap。华硕路由器内存虽然不小但Python解释器加上各种网络连接内存占用还是会慢慢爬升。强烈建议在USB盘上创建一个512MB的swap文件给系统一个缓冲余地。创建命令如下dd if/dev/zero of/opt/ai-gateway/swapfile bs1M count512 mkswap /opt/ai-gateway/swapfile swapon /opt/ai-gateway/swapfile开机自动挂载swap可以写到/jffs/scripts/post-mount这个钩子里在Entware挂载完成后执行swapon /opt/ai-gateway/swapfile。这个步骤能显著降低OOM杀进程的概率尤其是你同时还在路由器上跑着其他插件的时候。3. 提示流编排器核心设计3.1 一条请求的生命周期现在进入正题。理解这套系统的运行机制可以把一次完整的请求生命周期拆开看这是整个设计的骨架。第一步局域网内的某个设备向http://192.168.50.1:8787/v1/chat/completions发出一个POST请求请求头和OpenAI官方格式一致带着Authorization: Bearer开头的API Key。第二步服务端先校验API Key是否合法再校验来源IP是否在允许的网段内。这一步相当于门卫把陌生人挡在门外。第三步限流模块检查这个Key在当前时间窗口内是否已经刷了太多请求防的是某个应用死循环把云端API额度打爆。第四步从请求体里提取最后一条用户消息这是提示流处理的原料。第五步进入核心的提示流执行阶段。编排器读取config.json里的流程定义按顺序执行每个步骤。典型流程是先做规则路由根据关键词决定走哪个模型后端然后做模板渲染把用户消息嵌入到一个系统提示词里接着调用选定的后端模型接口拿到结果后做后处理比如截断长度、剥离多余换行。第六步如果启用了缓存系统会先查询缓存命中则直接返回结果不命中才调用后端。最后把模型返回的内容包装成OpenAI格式的响应写回给请求方。每一条请求都走这六步每一步都可以在配置里开关或调整。这种设计把“策略”和“实现”完全分离改行为不用动代码只改JSON这在家用路由器这种不方便登录调试的环境里是非常实用的特性。3.2 用JSON定义一条提示流配置是整个系统的灵魂。我把config.json写得尽量语义化让人一眼能看懂这条流水线的走向。贴一个实际在用的精简版{ server: { lan_ip: 192.168.50.1, port: 8787, api_keys: [edge-gw-2024], allowed_subnets: [192.168.50.0/24] }, rate_limit: { per_key_per_minute: 30 }, cache: { enable: true, ttl_seconds: 3600 }, backends: { home_ollama: { type: openai_compatible, base_url: http://192.168.50.10:11434/v1, model: qwen3:8b, timeout: 30 }, cloud_model: { type: openai_compatible, base_url: https://api.example.com/v1, model: gpt-4o-mini, timeout: 20 } }, flows: { default: { steps: [ { op: route, rules: [ { pattern: 天气|日历|提醒, backend: home_ollama } ], default_backend: cloud_model }, { op: prompt, template: 你是家庭AI助手。请用中文简洁回答控制在100字以内。用户问题{{ last_user_message }} }, { op: call_backend }, { op: trim, max_chars: 500 } ] } } }解释一下几个关键字段。backends定义了可选的模型后端目前统一用openai_compatible类型这意味着Ollama、vLLM、各类云端模型API只要是OpenAI格式都能接进来。timeout按后端单独设置本地模型给长一点云端服务短一点。flows.default.steps定义了默认提示流的四个步骤先根据正则规则选后端然后渲染模板接着调用后端最后修剪输出长度。这个配置有一个很实用的细节route里的pattern是正则表达式。如果用户输入包含“天气”或“日历”这类关键词请求会被路由到本地Ollama因为这类简单任务本地模型响应快、零成本。其他复杂任务走default_backend即云端模型。这样既省钱又分流是边缘网关的核心卖点之一。3.3 轻量网关的三张牌鉴权、限流、缓存一个网关如果只有转发功能那没有任何价值。这套系统在工程上之所以成立靠的是鉴权、限流、缓存这三张牌。鉴权做的是双层校验。第一层是API Key校验请求头里的Bearer token必须和config.json里的api_keys匹配。第二层是来源IP校验只有allowed_subnets里的网段才能访问服务。路由器上做IP校验有个天然优势它自己就是网关能明确知道请求是从哪个接口进来的。家里内网设备走LAN口访客网络走另一个网段可以在配置里只放行主网段让访客设备即使拿到Key也调不通。限流用的是内存滑动窗口。按API Key维度记录每分钟的请求时间戳超过阈值就返回429状态码。家庭场景默认每分钟30次已经非常宽裕但对异常程序来说这个限制能在云端账单爆炸之前踩刹车。实现的时候不需要Redis一个Python dict加一个deque就够用了毕竟这个网关只服务一个家庭局域网几十个并发就已经是极限场景。缓存是目前实际收益最大的一环。实现思路很直接把用户消息和流程名拼接后做SHA256哈希把哈希作为Key把模型返回结果和过期时间存进SQLite。TTL默认一小时。为什么这个环节收益大因为家庭成员对AI问的问题高度重复“今天有什么安排”这种话每天要说一遍第一遍花了3秒调用云端模型之后59分钟内的重复提问都是毫秒级返回。既省了云端API调用费又让设备端体验好了不少。4. 代码落地与部署实战4.1 核心代码入口服务项目核心代码分三个文件main.py负责HTTP入口flow.py负责提示流引擎cache.py负责缓存。main.py的核心结构非常精简标准库自带的http.server加ThreadingHTTPServer就能撑起整个服务。精简版代码大致长这样#!/usr/bin/env python3 import json import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer CONFIG json.load(open(/opt/ai-gateway/config.json, encodingutf-8)) SEMAPHORE threading.BoundedSemaphore(4) class GatewayHandler(BaseHTTPRequestHandler): def do_POST(self): if self.path ! /v1/chat/completions: self.send_error(404) return if not self._check_auth(): self.send_error(401) return length int(self.headers.get(Content-Length, 0)) body json.loads(self.rfile.read(length).decode(utf-8)) user_text body[messages][-1][content] with SEMAPHORE: result run_flow(default, user_text, CONFIG) payload json.dumps({ id: chatcmpl-edge, object: chat.completion, model: result[backend], choices: [{ index: 0, message: {role: assistant, content: result[text]}, finish_reason: stop }] }, ensure_asciiFalse).encode(utf-8) self.send_response(200) self.send_header(Content-Type, application/json; charsetutf-8) self.send_header(Content-Length, str(len(payload))) self.end_headers() self.wfile.write(payload) def _check_auth(self) - bool: token self.headers.get(Authorization, ) return token Bearer CONFIG[server][api_keys][0] if __name__ __main__: server ThreadingHTTPServer((CONFIG[server][lan_ip], CONFIG[server][port]), GatewayHandler) server.daemon_threads True server.serve_forever()重点说几个工程细节。信号量SEMAPHORE限制同一时间最多4个请求在跑这个数字是根据内存余量实测出来的。如果放开到几十个并发Python线程的内存开销加网络缓冲很容易把路由器的512MB内存吃穿。do_POST里先鉴权再解析body顺序不能反否则未授权请求也能触发JSON解析开销等于给了攻击者一个廉价的CPU消耗入口。响应体用的ensure_asciiFalse这个很重要否则中文会被转成\uXXXX的转义序列虽然客户端能正常解析但日志排查时看到全是一堆反斜杠头都要大。4.2 核心代码提示流引擎flow.py是整套系统的逻辑核心它负责把配置里的步骤一步步真正跑起来。核心函数run_flow的设计思路是遍历所有步骤用一个上下文ctx字典传递中间状态。贴关键片段def run_flow(flow_name: str, user_text: str, cfg: dict) - dict: steps cfg[flows][flow_name][steps] ctx {user_text: user_text} backend None for step in steps: op step[op] if op route: backend _route(step, user_text) ctx[backend] backend elif op prompt: ctx[prompt] _render(step[template], ctx) elif op call_backend: ctx[response] _call_backend(backend, ctx[prompt], cfg) elif op trim: ctx[response] ctx[response][: step.get(max_chars, 500)] return {text: ctx[response], backend: backend} def _route(step: dict, text: str) - str: for rule in step[rules]: if re.search(rule[pattern], text, re.I): return rule[backend] return step.get(default_backend, cloud_model) def _render(template: str, ctx: dict) - str: return template.replace({{ last_user_message }}, ctx[user_text])_route的逻辑很直白逐个规则做正则匹配第一个命中的后端返回。没有命中任何规则就落到默认后端。_render用的是最简单的字符串替换而不是Jinja2模板引擎这是故意这么做的路由器上没必要为一个模板功能引入几百KB的第三方库。_call_backend的实现用的是标准库的urllib.requestdef _call_backend(backend: str, prompt: str, cfg: dict) - str: conf cfg[backends][backend] url conf[base_url].rstrip(/) /chat/completions payload { model: conf[model], messages: [{role: user, content: prompt}] } req urllib.request.Request( url, datajson.dumps(payload).encode(utf-8), headers{Content-Type: application/json} ) with urllib.request.urlopen(req, timeoutconf.get(timeout, 30)) as resp: data json.loads(resp.read().decode(utf-8)) return data[choices][0][message][content]这个函数做的事情就是透传模型调用把渲染好的提示词发给后端然后从OpenAI格式的响应里取出文本返回。timeout从配置里读取本地Ollama给30秒云端模型给20秒超时时间必须单独控制否则一个慢后端会拖住整个网关的连接池。4.3 部署到路由器与自启动插件代码本地写完不是终点要让它成为路由器上的常驻服务还得做两步一是把文件传上去二是注册开机自启。上传我用的是SCP命令scp -r ai-gateway admin192.168.50.1:/opt/传完之后建议先在路由器上执行一遍语法检查cd /opt/ai-gateway python3 -m compileall .compileall能快速抓住缩进错误、语法错误这类低级问题比启动服务后在日志里翻报错高效得多。没问题之后手动跑一次start.sh看看能否正常拉起#!/bin/sh export PATH/opt/bin:/opt/sbin:/usr/sbin:/usr/bin:/sbin:/bin export PYTHONIOENCODINGutf-8 export LANGC.UTF-8 cd /opt/ai-gateway nohup python3 main.py logs/stdout.log 21 启动后立刻curl探一下端口确认服务已经起来了。开机自启我推荐用Entware的init脚本机制而不是直接写/jffs/scripts/services-start。原因很简单Entware有它自己一套服务管理流程在/opt/etc/init.d/目录下放一个S99aigw脚本如下#!/bin/sh PROCSmain.py ARGS/opt/ai-gateway/main.py DESCAI edge gateway这个脚本配合Entware的rc.common框架会在系统启动阶段自动被调用而且会在Entware挂载完成之后才执行避开了手动写services-start最头疼的时序问题。当然有些人更喜欢在jffs里加自己的启动钩子也不冲突只是要注意加sleep至少20秒给USB挂载和Entware启动留时间。还有一个防御手段是watchdog脚本。路由器上任何服务都有可能因为内存紧张被系统杀掉所以我另写了一个watchdog.sh每隔30秒检查一次进程是否存在不存在就重新拉起配合cru定时任务注册到系统cron里cru a aigw_watchdog */1 * * * * /opt/ai-gateway/watchdog.sh这样即使进程半夜被OOM杀掉最多一分钟内就能自动恢复不影响第二天早上的使用。4.4 局域网客户端接入实测部署完成后的验证环节我会分两个层面做测试。第一层是直接用curl发原始请求快速确认网关的鉴权、路由、调用链路是否正常curl -s http://192.168.50.1:8787/v1/chat/completions \ -H Authorization: Bearer edge-gw-2024 \ -H Content-Type: application/json \ -d {model:auto,messages:[{role:user,content:明天天气怎么样}]}如果配置的路由规则把“天气”归到本地Ollama这条请求会在一两秒内返回而且响应里能直接看到用的后端是home_ollama。第二层测试是用OpenAI的官方Python SDK接入这一层决定了家里其他设备接入是否方便。SDK只需要改base_url和api_key两个参数其他代码完全不用动from openai import OpenAI client OpenAI( base_urlhttp://192.168.50.1:8787/v1, api_keyedge-gw-2024 ) resp client.chat.completions.create( modelauto, messages[{role: user, content: 帮我把明天的待办整理成三点}] ) print(resp.choices[0].message.content)实测下来OpenAI SDK只需要这一个改动就能完整接入这意味着家里所有原本为云端API开发的脚本、Home Assistant的对话组件、甚至是某些开源聊天机器人的前端都能无缝切换到这个局域网网关。唯一要注意的是模型名填auto或者随便填网关不会真的把这个字段透传给后端后端选择完全由提示流的路由规则决定。5. 常见问题与排查实录5.1 服务起不来或开机丢失这是路由器自启服务最经典的问题。现象通常是刚部署完手动启动一切正常但重启路由器后服务就没了或者启动失败。排查方向很明确先看进程在不在再看日志有没有输出。第一步执行pgrep -f main.py如果没有输出说明进程压根没有拉起来。第二步检查/opt/bin/python3存不存在很多情况下重启后Entware还没挂载完成服务脚本已经跑了自然找不到解释器。第三种情况是启动脚本执行了但Python报错直接退出去logs/stdout.log里看有没有Traceback多半能定位到问题。最终的解决办法就是在自启动脚本里加足够的sleep等待或者改用Entware的init脚本机制来托管。5.2 路由器内存不足导致进程被杀如果你的路由器只有256MB内存这个坑几乎躲不掉。现象是进程运行一段时间后突然消失dmesg里能看到OOM killer的记录或者系统日志里出现“Out of memory”字样。我用free -m观察过Python进程加上内核缓存内存占用很容易攀升到接近上限。解决手段有三板斧。第一创建swap文件并确保开机自动启用给OOM killer一个缓冲。第二收紧服务端的并发信号量从4改成2虽然极限并发能力下降但内存增长会平缓很多。第三定期清理日志文件长时间运行的stdout.log会无脑膨胀在cron里加一条日志清理命令几行代码的事但能避免磁盘和内存双保险的压力。5.3 后端连接超时或返回异常网关本身没挂但设备端报错“connection error”或者“timeout”这时候要逐层排查。我的排查顺序是先越过网关直连后端确认模型服务本身活着再通过网关调一次看具体卡在哪一步。直连后端的命令很简单以本地Ollama为例在路由器上或者局域网内其他机器执行curl -s http://192.168.50.10:11434/v1/chat/completions \ -H Content-Type: application/json \ -d {model:qwen3:8b,messages:[{role:user,content:ping}]}如果直连返回正常说明问题出在网关和后端之间的超时配置。解决办法是调大config.json里对应后端的timeout值本地模型推理大上下文时确实需要更宽裕的时间。如果直连也不通检查后端服务的监听地址是否绑定在了正确的网卡上以及有没有防火墙拦截。云端API返回4xx错误的情况也遇到过。最常见的是模型名写错、上下文长度超限、或者某个必需参数缺失。因为网关是透传状态码的设备端看到什么错误直接按状态码去查原服务的文档就行。5.4 中文乱码与编码问题中文乱码的问题一半出在终端显示一半出在Python编码设置。如果你在SSH终端里手动启动服务然后发现终端打印的中文全是方块或者问号别急着改代码先检查终端软件自身的编码设置。我现在用的终端强制UTF-8编码专治路由器SSH会话里的中文显示问题。另一半的问题在服务本身。Python在写入日志、输出响应时默认编码可能不是UTF-8尤其是在Entware这种精简环境下。解决办法是在start.sh里加上两行环境变量export PYTHONIOENCODINGutf-8 export LANGC.UTF-8响应体的JSON序列化用ensure_asciiFalse这样返回的中文就是原始字符而不是转义序列。还有一个排查经验是日志里看到的乱码和实际API响应里的乱码没有必然联系客户端拿到什么数据要以实际抓包为准不要被SSH终端显示问题误导。最后分享一点实际体会这期项目前前后后折腾了两周踩过不少坑之后最大的一个体会是在资源受限的环境里做设计砍需求的能力比加功能的能力更重要。路由器上跑的服务必须把“配置驱动、单进程、零第三方依赖”当铁律来执行凡是有一条不满足都会在之后某个半夜变成让你爬起来救火的坑。也正因为有这些限制反而逼出了一个比服务器上跑的大体量方案更干净、更清晰的设计。目前这套网关在我家里已经稳定跑了一个多月下一步我打算用cru定时任务做一个早晨自动生成当日简报的流程再把它接进Home Assistant的对话组件里让路由器在家庭自动化里能再多管几件事。如果你有更好的想法也可以在这个思路上继续扩展。