企业官网建设流程全解析

周末我躺在沙发上，想远程看一眼那台主力机上跑着的 AI Agent 队列到底有没有卡住，结果发现除了回书房开电脑，我没有任何办法。这种感觉太憋屈了——我用 AI Agent 干了不少活，写周报、盯订单、抓数据分析，但这些全被绑死在"人必须在电脑前"这个前提上。花了一个周末，我用 FastAPI、LangGraph 和 Redis 把这套东西重做了一遍，现在是手机发条消息就能远程操控所有 AI Agent，干完自动把结果推回来。这篇文章就把整条链路完整拆开，从架构设计到并发处理，再到手机端实操，给同样想"把 Agent 揣进口袋"的人一份可直接抄的作业。

1. 为什么会有这个需求：本地 Agent 再强，人一离开电脑就抓瞎

1.1 我的真实使用场景

先说清楚我到底在跑什么 Agent，不然下面的方案全是空中楼阁。

我有三个常驻 Agent：第一个负责整理每天的工作日报，需要去拉数据库里的项目进度；第二个监控电商订单状态，发现有异常就自动标记并生成提醒；第三个是内容助手，帮我处理小红书上粉丝发来的私信，整理出待回复清单。这些任务有一个共同点：单次执行耗时长、需要调用多个外部工具，而且不是人坐在命令行前盯着一行行输出就能完事的。

以前我的操作模式是：在电脑上起一个终端，跑一个 Python 脚本，脚本内部调用 LangChain/LangGraph 编排好的 Agent，Agent 自己去决策要调哪些工具，最后把结果打出来。问题是，一旦我出门，这些 Agent 就像被关了禁闭——明明服务器还在跑，我却没有任何远程指挥和控制的手段。有一回我在外面，手机收到订单异常告警，愣是等到回家才处理。

所以我的核心需求不是"在手机上写代码"，而是把 Agent 的执行引擎留在服务器上，把遥控器放到手机里。手机上只需要做三件事：说清楚要干什么、查看任务跑到哪一步、拿到最终结果。

1.2 为什么不想完全依赖云端 Agent 平台

有人可能会说，你直接用扣子、Dify 这类现成的 Agent 平台不就完了吗？我也试过。说实话，这类平台在快速搭建 Demo 上确实爽，拖几个节点、配一下模型就能出一个能聊天的 Agent。但真到生产级使用，有几个问题很难绕过。

第一是工具链的绑定。我的订单监控要直连公司数据库，小红书私信要调用开放接口，这些私域数据源在云平台上配置起来非常别扭，要不就是权限模型对不上，要不就是网络打通困难。第二是并发策略没法自定义。平台给的是黑盒，底层排队、限流、worker 数量你说了不算，高峰时被限速了只能干瞪眼。第三是成本不可控。Agent 任务和普通 API 调用不同，一个任务会拆成多轮模型调用，按量计费在长时间任务下账单涨得很快。

自己做这套远程操控体系，本质上就是把"Agent 的执行权"和"Agent 的调度权"都攥在自己手里。我用的方案很简单：LangGraph 编排 Agent，FastAPI 包一层 HTTP 服务，Redis 做任务队列，手机端用 PWA 加快捷指令访问。整个过程完全自建，每一层都能控制，出了问题知道去哪查。

1.3 目标定义：把"遥控器"和"引擎"分开

动手之前我给自己定了个明确的目标，免得做着做着跑偏：

• 引擎（Agent 执行）必须跑在固定服务器上，不依赖手机端的算力
• 手机端必须是"无脑"的，界面上能发任务、能看状态、能收结果即可
• 任何操作都要有鉴权，毕竟手机丢了不是小事
• 要能扛住多人同时用，因为我后面想让团队也接入

这个定义非常重要。它决定了整个技术选型的方向：手机端不需要跑任何重型依赖，不需要装 Python 环境，只需要一个浏览器或者系统自带的快捷指令功能就够了。重型的东西全在服务端，包括 LangGraph 的状态图、Redis 队列、模型的调用限流。

2. 整体方案：Agent 服务端、消息网关、手机端三件套

2.1 服务端选型：FastAPI + LangGraph，以及为什么不折腾别的

服务端我选了三个核心组件，分工非常明确。

LangGraph 负责 Agent 本身的编排。它是 LangChain 生态里专门做复杂 Agent 流程的框架，跟普通链式调用不同，LangGraph 把 Agent 的每一步定义成图里的节点，比如"思考节点""工具调用节点""判断节点"，节点之间通过条件边跳转。这个设计特别适合真实世界的 Agent——不是一问一答，而是 Agent 可以自己在循环里决定下一步调哪个工具、要不要继续追问，直到它认为任务完成了才结束。

FastAPI 负责把 Agent 暴露成 HTTP 接口。选它是因为异步支持好、自带参数校验和接口文档，而且社区生态成熟，后面想加 WebSocket 推送结果也很方便。顺带说一嘴，热词里有人问"基于 Rust 语言能不能做 AI Agent"，当然可以，Rust 做出来的性能确实顶。但对绝大多数团队来说，Python 生态里 LangChain/LangGraph 的工具链、文档、示例是最完整的，开发效率远大于那点性能差距。Java 团队的话也可以参考 Spring AI，思路是一样的，只是实现语言不同。我自己选 Python 是因为业务迭代快，不太想把时间耗在生命周期管理上。

Redis 在这里不只是缓存，它是整个远程操控系统的"任务总线"。手机端发任务 -> 写入 Redis 队列 -> Worker 进程消费 -> 状态写回 Redis -> 手机端轮询读取。Redis 的列表结构天然支持队列，Hash 结构存任务状态，再加上可选的过期时间，轻量又好用。

2.2 手机端入口：PWA 控制台为主、IM 机器人为辅

手机端我做了两个入口，按推荐程度排序。

第一个是PWA 控制台，就是手机浏览器能直接打开、可以"添加到主屏幕"变成独立应用的那种网页。PWA 的好处是跨平台，iOS 和 Android 都能用，不用写原生 App，也不需要上架应用商店。我只需要写一个极简的 Web 页面，提供输入框（发任务）、任务列表（看状态）、结果展示区。页面通过 HTTPS 访问，配合 Service Worker，离线时还能缓存一部分静态资源。用系统自带的分享功能或者快捷指令调起这个 PWA，体验接近原生 App。

第二个是IM 机器人，比如企业微信群机器人或者钉钉机器人。这类入口的好处是消息本身就是任务输入，结果也能直接以消息形式回推到群里，适合"一句话触发任务"的场景。不过实现和维护成本比 PWA 高一点，要考虑消息格式解析、回调地址配置、主动推送权限这些。我的建议是先把 PWA 跑通，再按需加 IM 入口。

我当时坚持要两个入口同时能用，原因很简单：PWA 适合"坐下来慢慢看"的场景，IM 机器人适合"顺手发一句就完事"的场景。俩者共用同一套后端 API，不存在双份维护的问题。

2.3 通信协议：任务 ID、状态机与轮询模型

手机和服务器之间怎么通信？这是整套系统里最容易想当然、最容易翻车的地方。

最朴素的想法是：手机发个请求，服务器当场把 Agent 跑完，再把结果返回。看起来没什么问题，但实际一跑就露馅——Agent 任务动辄几十秒甚至几分钟，HTTP 连接根本撑不了那么久，中间网络稍微抖一下整个请求就断了，手机那边以为自己没发出去，服务器那边其实已经跑了一多半。

我改成了任务制：手机端发任务请求时，服务端只负责生成一个任务 ID，把任务写进 Redis 队列，然后立刻返回"任务已接收，ID 是 xxx"。真正执行 Agent 的是后台的 Worker 进程，它从队列里取任务、执行、把进度状态和最终结果不断写回 Redis。手机端拿着任务 ID 去轮询，每隔两三秒查一次状态，直到状态变成 done 或者 error。

背后的状态机很简单：

这个模型最大的好处是解耦。手机发完任务就断网也不怕，任务在服务器上照跑，等手机恢复了再查状态就行。后面要接 IM 机器人、短信通知，都是在这个状态机上扩展通知逻辑。

3. 核心搭建：把 LangGraph Agent 变成远程服务

3.1 先写一个会"思考-调用-再思考"的 Agent

先建一个最简但完整的 LangGraph Agent，让它具备"理解用户请求 -> 决定调用工具 -> 拿到工具结果 -> 继续推理"的闭环能力。我用的是两个工具：查订单状态和查天气，实际场景替换成你自己的数据源就行。

# agent_core.py from typing import TypedDict from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import ToolMessage class AgentState(TypedDict): messages: list next: str @tool def query_order(order_id: str) -> str: """查询指定订单的最新物流状态。""" # 这里替换成你自己的数据源，比如 MySQL 查询或者调用第三方 API return "订单{}当前状态：已发货，预计三天内送达。".format(order_id) @tool def get_weather(city: str) -> str: """查询指定城市的实时天气。""" return "{}今天晴，气温28度，微风。".format(city) tools = [query_order, get_weather] llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) llm_with_tools = llm.bind_tools(tools) def agent_node(state: AgentState) -> dict: response = llm_with_tools.invoke(state["messages"]) return {"messages": state["messages"] + [response]} def tool_node(state: AgentState) -> dict: messages = state["messages"] last = messages[-1] new_messages = [] for call in last.tool_calls: tool_name = call["name"] args = call["args"] if tool_name == "query_order": result = query_order.invoke(args) elif tool_name == "get_weather": result = get_weather.invoke(args) else: result = "未知工具" new_messages.append(ToolMessage(content=str(result), tool_call_id=call["id"])) return {"messages": messages + new_messages} def should_continue(state: AgentState) -> str: last = state["messages"][-1] if getattr(last, "tool_calls", None): return "tools" return END graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", tool_node) graph.set_entry_point("agent") graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END}) graph.add_edge("tools", "agent") remote_agent = graph.compile()

这段代码的核心逻辑在should_continue这个判断上。模型每次输出的消息里如果带了 tool_calls，说明它想调用工具，那就跳进工具节点；如果模型觉得问题已经被回答了，没有工具调用，就直接结束。这个过程会循环，直到模型不再想调用工具为止。iPhone 用户可以把这个过程类比成"你和 ChatGPT 来回对话多轮，每轮它都可能说'我需要先查一下资料'"。

3.2 FastAPI 接口与任务下发

接下来把 Agent 包成 HTTP 服务。这里最关键的思维转变是：不能让请求处理函数直接执行 Agent，它只负责登记任务。

# server.py import uuid from fastapi import FastAPI, HTTPException, Depends, Header from pydantic import BaseModel from redis import Redis app = FastAPI() redis = Redis(host="127.0.0.1", port=6379, decode_responses=True) API_TOKEN = "your-secret-token" async def verify_token(x_token: str = Header(default="")): if x_token != API_TOKEN: raise HTTPException(status_code=401, detail="invalid token") class RunRequest(BaseModel): prompt: str @app.post("/agent/run", dependencies=[Depends(verify_token)]) async def run_agent(req: RunRequest): task_id = str(uuid.uuid4()) # 任务描述入队 redis.rpush("agent:queue", f"{task_id}|{req.prompt}") # 初始化任务状态 redis.hset(f"agent:task:{task_id}", mapping={ "status": "queued", "prompt": req.prompt, "result": "", "error": "" }) redis.expire(f"agent:task:{task_id}", 3600 * 24) return {"task_id": task_id, "status": "queued"} @app.get("/agent/task/{task_id}", dependencies=[Depends(verify_token)]) async def get_task(task_id: str): data = redis.hgetall(f"agent:task:{task_id}") if not data: raise HTTPException(status_code=404, detail="task not found") return data

注意几个细节。

redis.hset里我用了一个 Hash 来存任务的所有元信息，包括状态、原始请求、结果、错误信息。这样手机端一次查询就能拿到全部需要的数据。redis.expire给任务设置了 24 小时过期，防止 Redis 里堆积太多陈旧任务占用内存。

为什么要把 prompt 也存一份？因为轮询接口 http://your-server/agent/task/xxx 返回的数据里要回显用户最初发的消息，这样才能在手机端画出一个"我问了什么、Agent 答了什么"的完整对话视图。如果不存，任务列表里只有结果没有原问题，体验会很奇怪。

鉴权我用了一个简单的 X-Token 头，通过 FastAPI 的依赖注入统一校验。生产环境建议升级成 JWT 或者按用户维度分配 Token，后面在安全章节细说。

3.3 Worker 消费队列：真正执行 Agent 的地方

服务端把任务丢进队列就完事了，真正费时费力的工作由 Worker 进程来干。Worker 的逻辑用死循环实现，不断从 Redis 队列里取任务，取到了就更新状态为 running，然后调用 remote_agent 执行，执行完把结果写回 Redis。

# worker.py import time from redis import Redis from agent_core import remote_agent redis = Redis(host="127.0.0.1", port=6379, decode_responses=True) while True: item = redis.blpop("agent:queue", timeout=5) if not item: continue task_id, prompt = item[1].split("|", 1) redis.hset(f"agent:task:{task_id}", "status", "running") print(f"[{time.strftime('%H:%M:%S')}] task {task_id} started, prompt={prompt}") try: result = remote_agent.invoke({"messages": [{"role": "user", "content": prompt}]}) answer = result["messages"][-1].content redis.hset(f"agent:task:{task_id}", mapping={ "status": "done", "result": answer }) print(f"[{time.strftime('%H:%M:%S')}] task {task_id} done") except Exception as e: redis.hset(f"agent:task:{task_id}", mapping={ "status": "error", "error": str(e) }) print(f"[{time.strftime('%H:%M:%S')}] task {task_id} error: {e}")

这里的blpop是 Redis 的阻塞弹出命令，队列为空时它会一直等，而不是轮询空转。这个细节很重要——如果换成lpop+ sleep，Worker 会在没任务时白白烧 CPU；而blpop可以让 Worker 彻底休眠，等队列里有数据了立刻被唤醒。

Worker 进程和 API 进程是完全独立的两套进程。我跑的时候 API 用 uvicorn 启动，想开几个进程就开几个；Worker 用同样的方式在后台挂着，想提高处理能力就多起几个 Worker 进程。这两者之间的通信完全通过 Redis，互不干扰。有一回 Redis 临时挂了，API 照样能启动，只是任务进不了队，Worker 全在空转，排查起来特别清晰。

3.4 状态查询与中间结果

很多 Agent 任务不只是"最后给个结果"那么简单，中间过程本身就有价值。比如内容助手在整理小红书私信时，可能会阶段性地产生"已汇总 12 条未读私信，正在按紧急程度排序"这种中间状态。手机端用户看到进度，就知道 Agent 没死，正在干活。

这个功能怎么加？其实只需要在 Worker 执行 Agent 的过程中，往任务 Hash 里多写一个progress字段。LangGraph 的节点函数里可以随时更新外部状态，我在agent_node里调了模型之后、工具之前，把当前模型输出追加到 Redis 的 progress 字段里。

# 在 agent_node 里（注意这是 Worker 进程内） def agent_node(state: AgentState) -> dict: response = llm_with_tools.invoke(state["messages"]) if getattr(response, "tool_calls", None): tool_names = ", ".join([c["name"] for c in response.tool_calls]) redis.rpush(f"agent:task:{task_id}:progress", f"计划调用工具：{tool_names}") return {"messages": state["messages"] + [response]}

手机端轮询时把 progress 列表一起拉回去，渲染成类似"第 1 步：计划调用工具"这样的时间线。尽管不是严格意义上的流式输出，但已经足够让用户感知到任务的推进，而不是干等着那一个光标转圈。这一步对远程操控的体验提升非常明显。

4. 并发才是真难点：手机背后的人潮会让 Agent 现出原形

4.1 Agent 任务为什么比普通接口重得多

热词榜上挂着的"ai agent 怎么扛并发"，我在这上面栽过跟头，所以单独开一章说。

普通 Web 接口给用户的体感是毫秒级或者秒级返回，就算后端逻辑复杂，也能通过加机器、加线程池硬扛。Agent 任务完全不是这个量级——一次任务的耗时不是百毫秒，而是几十秒到几分钟。假设一个 Agent 任务平均耗时 60 秒，你在家里起 4 个 Worker 进程，意味着每分钟最多只能完成 4 个任务。如果手机端有 10 个人同时每人提交 2 个任务，队列立刻积压，而且积压量只增不减。

更隐蔽的问题是模型 API 的速率限制。LangGraph 里每次和模型对话都是一次 API 调用，而一个 Agent 任务可能要和多轮模型 API 打交道。就算你起了 20 个 Worker，模型提供商那边可能只允许你每分钟调用 30 次。Worker 再多，也会被模型 API 端的 RPM/TPM 限制卡住。我第一次并发测试时就是这么挂的，看起来 Boom 了 20 个 Worker，实际全被 429 限流错误打爆。

所以扛并发的本质不是"提高单机吞吐"，而是控制任务的生产速度和消费速度，让系统在水位线以内稳定运行。

4.2 队列、Worker 池、限流三件套

我的解法是三层结构：Redis 队列承受突发流量，Worker 池决定消费速率，令牌桶限制模型 API 调用频率。

第一层，Redis 列表当队列。手机端提交任务时，不管同时来多少请求，Redis 的 rpush 都是原子操作，不会丢数据。队列天然具备削峰填谷的作用——瞬时 50 个任务进来，就先堆在队列里，Worker 按自己的节奏一个一个吃。

第二层，Worker 池的规模要和模型 API 的限流匹配。不要无限开 Worker，开多了只会制造一堆报 429 的日志。我用一个非常土但是有效的办法：先看模型提供商给的 RPM 上限，比如每分钟 60 次；一个 Worker 每执行一个任务平均要调 5 次模型 API，那每分钟最多消耗 60 次调用。理论上 Worker 数 = 60 / 5，也就是最多 12 个。实测建议先 4 到 6 个起步，观察队列消费速率再调整。

第三层，令牌桶限流。即便 Worker 数大约摸对了，模型 API 的返回时间也有波动，短时间内的并发调用量可能突刺。我在 Worker 的模型调用处加了一个简单的令牌桶，每秒放行一定数量的请求，超出的让它们排队等。

# 简化的令牌桶实现 import time class TokenBucket: def __init__(self, rate_per_second, capacity): self.rate = rate_per_second self.capacity = capacity self.tokens = capacity self.last = time.monotonic() def acquire(self, block=True): while True: now = time.monotonic() self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate) self.last = now if self.tokens >= 1: self.tokens -= 1 return True if not block: return False time.sleep(0.05)

用的时候在 Worker 调模型 API 之前bucket.acquire()一下，把全局调用频率限制在设定值以内。别小看这个土办法，它彻底解决了 429 问题。

4.3 我压测下来的真实数据与参数调优

最后给一组我自己机器上的实测数据，配置是 4 核 CPU、16GB 内存、Redis 和 Worker 都在这台机器上。

结论有两个：一是 Worker 数量真不是越大越好，尤其模型 API 有限流时，多余的 Worker 全在做无用功；二是队列有积压并不可怕，怕的是积压增长速度大于消费速度，48 小时都清不完。实际运营中我给手机端还加了一层前置的提交频率限制，同一个手机号每分钟最多提交 3 个任务，避免手滑连点把队列打爆。

补充一句关于"ai agent 中台"的想法：如果你不想自己维护这套底层，可以考虑把队列、Worker、模型限流这套封装成公司内部的中台服务，业务方只需要往队列里塞任务，不用关心 Agent 怎么跑。这个方向有点重，但团队用到 3 个人以上时很值得投入。

5. 手机端实操：PWA 控制台 + 快捷指令上手

5.1 手机浏览器里的 PWA 控制台

手机端的核心是一个纯前端的 PWA 页面，不需要任何 App 壳子，本质上就是一个静态网页，所有数据来自后端 API。我写了一个极简版本，核心功能包括：输入任务、展示任务列表、轮询状态并渲染结果。

页面的骨架长这样：

<!doctype html> <html> <head> <meta charset="utf-8" /> <meta name="viewport" content="width=device-width, initial-scale=1" /> <title>Agent 遥控器</title> <link rel="manifest" href="/manifest.json" /> </head> <body> <h1>Agent 遥控器</h1> <input id="prompt" placeholder="描述你想让 Agent 干的事" /> <button id="send">下发任务</button> <div id="tasks"></div> <script src="/app.js"></script> </body> </html>

manifest.json里配置一下应用名称、图标、主题色，手机浏览器就会在"添加到主屏幕"时把它当成一个独立应用。这个文件很容易被忽略，但如果没有它，PWA 只是"网页快捷方式"，体验差一截。

JavaScript 逻辑里有两个核心函数。发送任务用 fetch POST/agent/run，拿到 task_id 之后立即加入前端列表，并开启一个 setInterval 每 3 秒轮询一次该任务的状态；状态为 done/error 时停止轮询并渲染结果。3 秒这个间隔是我调出来的平衡值，太密容易把服务器打得太热，太稀任务完成通知不够及时。

async function sendTask() { const prompt = document.getElementById('prompt').value; const resp = await fetch('/agent/run', { method: 'POST', headers: { 'Content-Type': 'application/json', 'X-Token': localStorage.getItem('token') || '' }, body: JSON.stringify({ prompt }) }); const data = await resp.json(); pollTask(data.task_id); } async function pollTask(taskId) { const resp = await fetch(`/agent/task/${taskId}`, { headers: { 'X-Token': localStorage.getItem('token') || '' } }); const data = await resp.json(); renderTask(data); if (data.status === 'running' || data.status === 'queued') { setTimeout(() => pollTask(taskId), 3000); } }

Token 我存在 localStorage 里而不是每次手输，手机端用起来更顺滑，但代价是丢了手机相当于丢了 Token，所以我给 Token 设置了轮换机制，服务端支持多 Token，发现异常随时吊销。

5.2 用 iOS 快捷指令一键触发 Agent

PWA 适合打开页面慢慢看，但如果我在路上只想说一句话就让 Agent 干活，打开网页再打字还是太慢。iOS 的快捷指令（Shortcuts）提供了一个非常顺手的方案。

我建了一个快捷指令"指挥 Agent"，流程是：要求输入文本 -> 用"获取 URL 内容"动作 POST 到/agent/run-> 解析返回的 JSON 里的 task_id -> 再用"获取 URL 内容"GET/agent/task/{task_id}轮询三次 -> 最终把结果读出来播报。整个过程完全不需要打开浏览器。

快捷指令的 URL 内容长这样：

POST https://your-server.com/agent/run Headers: X-Token: your-token, Content-Type: application/json Body: {"prompt": "刚才输入的文本"}

然后把它添加到 iPhone 的主屏幕或者用"连按两下背面"触发。Android 上也可以把快捷方式做成桌面小组件。这个方案用起来是真的爽，我现在更常用的是快捷指令，PWA 反而成了"看历史记录和详细结果"的工具。

5.3 结果通知与"长任务离线等结果"体验

最后一个体验上的加分项是任务完成通知。既然是远程操控，用户不可能一直开着页面盯着。做完任务之后怎么通知？

最简单的方案就是让 PWA 的 Service Worker 在后台轮询接口，一旦发现任务状态变成 done，就用系统通知 API 弹一条"任务完成"推送。Service Worker 的注册代码如下：

if ('serviceWorker' in navigator) { navigator.serviceWorker.register('/sw.js'); }

然后在 sw.js 里做后台任务检查，发现 done 状态就触发self.registration.showNotification。系统通知的标题和正文可以带上任务的原始描述，用户体验非常接近原生推送。

这里有个坑要提醒：Service Worker 在 iOS Safari 上的后台执行非常受限，锁屏之后基本没法持续轮询。我实测下来，Android Chrome 上表现还行，iOS 上最靠谱的其实是走"服务器 -> IM 群机器人 -> 用户手机"的链路。所以我群里也挂了个机器人，每次任务完成自动 @ 我。前面说 PWA 是主入口，这是从使用频率上说的，真要保证"任何情况都能收到通知"，服务端主动推消息的 IM 机器人反而更可靠。

6. 部署安全与第四级延伸：从自用工具到轻量中台

6.1 公网访问和 HTTPS 的正确姿势

手机要远程访问，服务就必须暴露在公网上。我直接把服务部署在云服务器上，采用 Nginx 反向代理 + HTTPS 证书的标准姿势。

Nginx 配置的核心片段如下：

server { listen 443 ssl; server_name agent.example.com; ssl_certificate /etc/nginx/ssl/agent.crt; ssl_certificate_key /etc/nginx/ssl/agent.key; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_read_timeout 60s; } }

几个关键点：证书用免费的 ACME 协议申请，不用自己掏钱买；proxy_read_timeout要设置合理，因为任务提交接口很快返回，这个值不用太长，但别设成 5 秒那种默认值，后面接流式接口时容易被断；Nginx 层面同时挂上速率限制，每个 IP 每秒钟最多 10 个请求，能挡住一部分扫描和滥用。

如果你不想买云服务器，也可以用内网穿透的方式。家里有台机器跑 Agent，在外面通过隧道访问，效果类似，但这块涉及具体的网络环境配置，我建议还是优先走正规云服务。

6.2 密钥、Token、工具权限的分层保护

这是远程操控方案里最不能省的一环，因为手机一旦丢了或者 Token 泄露，别人就能指挥你的 Agent 干任何事。我的安全设计分三层。

第一层是 API Token。每个设备分配一个独立的 Token，Token 在 Redis 里维护，可以单独吊销。手机端请求时必须带 X-Token 头，验证不通过一律 401。PWA 里我做了针对 Token 的过期提醒，30 天强制换一次。

第二层是工具白名单。LangGraph 里每个工具都检查调用者身份，在工具执行前加一道判断。比如内容助手的小红书私信工具，只有来自我绑定的手机 Token 的任务才允许执行，其他来源的任务直接拒绝。这样就算 API Token 被偷了，坏人也不能用你的 Agent 干出格的事。

第三层是危险操作的复核机制。大家都知道"个人使用 AI Agent 可以做期货交易吗"之类的讨论，我的态度是：跟钱强相关、会产生对外影响的操作，一律不自动执行。在我的体系里，Agent 可以分析行情、拉取数据、生成建议，但下单这种动作必须经过我在手机上二次确认。实现上就是一个带 confirm 标志的任务类型，Worker 执行到关键工具前停下，等用户在 PWA 上点一下确认才继续。

6.3 如何扩展成团队可用的 AI Agent 中台

如果你用了一两周觉得这套东西顺手，团队里其他人也说想用，就可以往轻量中台的方向演进。我目前已经在做的扩展有四个：

• 用户体系：从单 Token 变成账号密码登录，Agent 任务带上 user_id，互相隔离
• 审计日志：所有任务请求、工具调用、结果输出全部落库，方便回溯问题
• 工具注册中心：把工具从代码硬编码改为配置化注册，新工具通过后端页面添加即可，不用改 Worker 代码
• 限流策略细化：不同用户组分配不同的队列优先级和模型 API 配额，普通用户每天 50 次任务，运营组可以 500 次

这层扩展技术上不复杂，核心改动在数据模型上，把 Redis 里孤立的任务变成按用户维度管理，再加一层数据库存储历史记录。这个演进路径我走了三个月，每一步都是被实际问题逼出来的，没有一步是为了追求技术复杂度。

7. 个人体会：这套系统真正改变的是什么

动手做完这套手机远程操控 AI Agent 的系统，我最直观的感受是：AI Agent 终于从"电脑上的玩具"变成了"随身工具"。以前它是被拴在办公桌上的，现在它更像是我的一个远程同事——我给它布置任务，它干活，干完了主动汇报。我自己在建设过程中踩过的最大一个坑就是一开始试图让 HTTP 接口同步等 Agent 跑完，结果一连串超时和失败，后来改成任务队列模式，整个世界清净了。

最后分享一个小技巧：所有的轮询逻辑，无论 PWA 还是快捷指令，不要用固定的 1 秒短轮询，至少给 3 到 5 秒的间隔。一方面减少服务器压力，另一方面轮询太密集反而会在 Agent 任务还未完成时浪费时间重试。把任务状态机设计清楚、把结果缓存做好，比无限缩短轮询间隔有效得多。这套方案的花费只是一台低配云服务器加一个模型 API Key，动手成本不高，但带来的自由度提升是实打实的——现在我在任何地方，只要掏出手机，就能指挥我那一群 Agent 干活了。