企业官网建设流程全解析

直接说结论：这事儿可行，而且不需要多高级的设备，一台手机就能把服务端那一堆 AI Agent 全管起来。我花了一个周末把家里的工作流改成了这套“手机遥控器”模式，白天在外面用手机下发任务、看进度、拿结果，回家只需要处理 Agent 跑出来的异常和需要人工拍板的节点。整套方案的技术栈就是 FastAPI + LangChain + LangGraph，再加一个轻量级任务队列，移动端只做指令下发和状态展示。下面把从思路到踩坑的完整过程都写出来，想复制的直接照抄，想魔改的也可以顺着我的思路自己展开。

1. 先搞清楚你要解决什么问题

1.1 手机不是“算力中心”，而是“遥控器”

很多人一听“手机远程操控 AI Agent”，第一反应是把大模型跑在手机上，或者把 Agent 调度系统塞进手机里。说实话，这方向在消费级硬件上根本走不通：参数稍微大点的模型，手机内存和算力直接爆表；就算能跑，发热和耗电也会让体验变得非常糟糕。

我最后定的思路是：手机只负责发指令和收结果，所有 Agent 的算力、模型调用、工具执行全部留在服务端。这就跟遥控器控制空调一样——你手里拿的是面板，真正制冷制热的压缩机在室外机里。这样手机端的压力几乎为零，一张 Web 页面就能搞定，甚至不需要装 App。

这套设计解决的实际问题很明确：我不可能随时随地都坐在电脑前守着 Agent 跑任务，但通勤路上、午休时间、外出办事的时候，手机是唯一能随手拿起来的设备。把 Agent 的控制权搬到手机上以后，相当于把整套自动化体系变成了“口袋里的调度台”。

1.2 方案选型：从技术栈到部署方式

我最终选定的核心组合是FastAPI + LangGraph，再配一个本地任务队列和一份简单的 JSON 消息协议。选型逻辑后面会详细说，这里先讲整体方向：

• 服务端提供 HTTP API，接收手机端下发的任务指令，返回任务 ID 和状态。
• Agent 内部流程用 LangGraph 编排，写清“先做什么、后做什么、做不了怎么办”。
• 任务执行放进队列，避免手机端一个请求就把服务拖垮。
• 手机端用浏览器打开一个极简管理页，本质上就是几个按钮加一张任务列表。

部署方式上，我用了一台云主机跑服务端，再通过 HTTPS 暴露管理入口。云主机的好处是稳定、公网可达、不用折腾家里网络的动态 IP 问题。Agent 涉及的工具调用、模型 API 调用也都能在主机上顺畅跑通，手机这边全程只做轻量访问。

2. 架构设计：可被远控的 Agent 系统长什么样

2.1 我最终确定的系统架构

整套系统拆成四层，每层只干一件事：

手机端发一个“帮我写一篇关于 X 的文章大纲”这种任务，请求到 API 层以后，API 层不会立刻跑 Agent，而是先创建一个任务记录，丢进队列，返回一个task_id。后台 Worker 发现队列里有新任务，才真正把任务交给 LangGraph 去执行。手机端拿着task_id轮询状态接口，就能看到“排队中”“执行中”“执行完成”这些状态。

这个“任务不阻塞请求”的设计非常关键。Agent 一个任务可能跑几分钟甚至几十分钟，HTTP 请求不可能一直挂着。如果直接同步执行，手机端一个请求出去就卡住了，超时、断网、误操作这些不可控因素会把体验拉到最低。改成异步以后，丢任务和看结果完全分离，手机端体验接近“发个微信消息然后等回复”的流畅度。

2.2 为什么用 FastAPI 而不是 Django / Spring

不少人问过我：Django 那么成熟，Spring AI Agent 也很火，为什么不选它们？我的判断很简单——选型不是比谁功能多，而是比谁更贴合场景。

Django 确实生态丰富，自带 Admin 后台、ORM、认证体系，但它的异步支持相对晚，默认的同步模型在高并发 I/O 场景下有点吃劲。我的场景是大量短请求（手机端频繁轮询状态）+ 少量长任务（Agent 实际执行），FastAPI 原生异步接口处理这种 I/O 密集型的杂活非常顺手，性能表现也稳。

Spring 系适合 Java 技术栈成熟的团队，如果你本身是 Java 生态，用 Spring AI Agent 也没问题，但我的 Agent 编排、模型调用、工具封装全在 Python 生态里，硬切 Java 等于把已经熟练的路推倒重走。Rust 写的 Agent 性能确实强，但团队学习成本高、迭代速度慢，不是一个人玩转自动化工作流的友好选择。

FastAPI 还有一个隐性优势：自动生成 OpenAPI 文档。手机端联调的时候，直接打开/docs页面就能看到所有接口的定义和参数，省了一堆来回沟通的时间。

2.3 用 LangGraph 编排 Agent，而不是自己堆 if-else

早期写 Agent 的时候，我习惯在代码里写一大串if判断来决定下一步调什么工具，这种写法在流程固定时没问题，一旦出现分支多、需要人工介入、有循环重试的情况，代码马上变成一团乱麻。

LangGraph 解决的核心问题是把流程变成一张可执行的图。每个节点是一个动作（比如“搜索信息”“调用模型生成文案”“整理结果”），边定义了动作之间的流转条件。这样有两个直接好处：

• 流程可见。跑完一个任务，我能看到它走了哪些节点，卡在哪个环节，改了哪条边影响多大。
• 流程可控。可以给节点加回调、超时、重试，甚至接消息通知，把异常干预做到每一个环节。

而且 LangGraph 天然适配 LangChain 生态，模型调用、工具封装都是现成的，省去自己造轮子的过程。

一个手机远控场景，最怕的就是 Agent 内部变成一个黑盒——手机端只看到“执行中”，但不知道里面在干嘛。LangGraph 的节点状态机制让我可以把“当前在哪个节点、上一步结果是什么”透传出去，手机上能看到任务推进的过程，比自己写一堆日志再解析靠谱得多。

3. 手机端怎么“反控”服务端的 Agent

3.1 消息协议设计：让手机和 Agent 说同一种语言

手机端和服务端之间的通信，我做了一份特别简单的 JSON 协议，核心就三个接口：

• 创建任务：POST /tasks，请求体里包含任务类型和参数。
• 查询状态：GET /tasks/{task_id}，返回状态、进度、结果。
• 查看可用 Agent：GET /agents，展示当前服务端注册了哪些能干活的工作流。

任务请求体的格式参考：

{ "agent_id": "content_writer", "action": "generate_outline", "params": { "topic": "如何用手机管理AI Agent", "style": "专业口语化", "depth": 3 }, "callback": "optional_webhook_url" }

agent_id用来告诉服务端调哪个 Agent 工作流，action是具体动作，params是参数，callback是可选的回调地址。协议设计得越简单越好，因为手机端没有那么多复杂逻辑，能发 JSON 和解析 JSON 就够了。

这里有一条经验：协议字段宁少勿多。我一开始设计了十几个字段，后来发现真正用到的就五六个，多余的字段不仅让文档难写，手机端参数校验也麻烦。简约协议配合清晰的字段命名，长期维护成本最低。

3.2 状态接口：让手机随时知道 Agent 在忙什么

任务状态我用一个简单的枚举来描述：

手机端每隔两三秒轮询一次状态接口，拿到对应状态做展示。轮询看起来笨，但对这种低频控制场景来说最省事，不需要 WebSocket，也不需要维护长连接，一台手机同时盯几个 Agent 绰绰有余。

当任务跑出结果，服务端会返回一串结构化的数据，手机端直接渲染成卡片。比如让 Agent 写文章大纲，返回的可能是标题、分节、每节要点；让 Agent 分析数据，返回的可能是结论和图表数据的 JSON。手机端不需要理解内容本身，只需要按预设的模板展示。

3.3 身份验证与安全：不是任何设备都能发指令

手机远程下发指令，意味着服务端暴露在公网上，必须做身份验证，否则任何人拿到地址都能操控你的 Agent。我用的方案很朴素但很有效：

• 在 FastAPI 里加了一个简单的Authorization: Bearer <token>校验。
• 一份长期 API Token 给手机端管理后台用，Token 本身存手机浏览器里，不写死在页面代码中。
• 所有的 Agent 执行行为都限定在提供方配置的权限范围内，不能让遥控者通过 Agent 越权操作系统。

安全配置的底线意识还是要有的：宁可牺牲一点便利性，也不能把管理入口裸奔在公网上。

4. 核心代码与实操过程

4.1 搭建 FastAPI 服务骨架

我的 FastAPI 服务端代码非常规整，核心就是一个创建任务的接口和一个查询状态的接口。直接贴一份简化后的代码，隐藏掉业务细节，框架结构大家可以参考：

# main.py from fastapi import FastAPI, HTTPException, Header from pydantic import BaseModel from typing import Optional, Any import uuid import asyncio app = FastAPI(title="Mobile Agent Controller") # 任务请求体 class TaskRequest(BaseModel): agent_id: str action: str params: dict = {} callback: Optional[str] = None # 用一个字典模拟任务存储 TASK_STORE = {} VALID_TOKENS = {"your_long_random_token_here"} def check_auth(authorization: str): token = authorization.replace("Bearer ", "").strip() if authorization else "" if token not in VALID_TOKENS: raise HTTPException(status_code=401, detail="Invalid token") @app.post("/tasks") async def create_task(task: TaskRequest, authorization: str = Header(default="")): check_auth(authorization) task_id = str(uuid.uuid4()) TASK_STORE[task_id] = { "task_id": task_id, "status": "queued", "agent_id": task.agent_id, "action": task.action, "params": task.params, "result": None, "progress": [], } # 丢入后台任务队列（此处用 asyncio.create_task 演示异步执行） asyncio.create_task(run_agent(task_id, task)) return {"task_id": task_id, "status": "queued"} @app.get("/tasks/{task_id}") async def get_task(task_id: str, authorization: str = Header(default="")): check_auth(authorization) if task_id not in TASK_STORE: raise HTTPException(status_code=404, detail="Task not found") return TASK_STORE[task_id] async def run_agent(task_id: str, task: TaskRequest): # 模拟 Agent 执行，可以替换为 LangGraph 工作流调用 TASK_STORE[task_id]["status"] = "running" TASK_STORE[task_id]["progress"].append("start running") try: await asyncio.sleep(3) # 模拟耗时操作 TASK_STORE[task_id]["result"] = { "message": f"Finished {task.action} with params {task.params}" } TASK_STORE[task_id]["status"] = "completed" except Exception as e: TASK_STORE[task_id]["status"] = "failed" TASK_STORE[task_id]["result"] = {"error": str(e)}

这段代码里asyncio.create_task只是最简单的演示。生产环境里我建议把任务丢给真正的队列（比如 Celery 或者简单的 Redis 队列），因为asyncio.create_task在进程重启后会丢任务，队列持久化才经得起折腾。

4.2 把 LangGraph 工作流接进任务执行

FastAPI 的接口负责接单，真正干活的还是 LangGraph 工作流。我定义了一个非常典型的内容生成 Agent，包含“接收主题 -> 搜索资料 -> 生成大纲 -> 检查并返回”四个节点。

# agent_workflow.py from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): topic: str search_results: List[str] outline: str step_count: Annotated[int, operator.add] def search_node(state: AgentState): # 模拟搜索资料，可以替换为真实的搜索 API 调用 state["search_results"] = [f"result about {state['topic']}"] state["step_count"] = 1 return state def generate_outline_node(state: AgentState): # 调用大模型生成大纲，这里用字符串模拟 state["outline"] = f"这是关于{state['topic']}的文章大纲" state["step_count"] = 1 return state def review_node(state: AgentState): # 校验并返回 if "大纲" in state["outline"]: state["step_count"] = 1 return state else: return state graph = StateGraph(AgentState) graph.add_node("search", search_node) graph.add_node("generate_outline", generate_outline_node) graph.add_node("review", review_node) graph.set_entry_point("search") graph.add_edge("search", "generate_outline") graph.add_edge("generate_outline", "review") graph.add_edge("review", END) app_workflow = graph.compile()

这个工作流跑起来以后，我可以在手机端看到每一步的进度：开始搜索、生成大纲、校验返回。如果想让手机端实时看到当前节点，只需在 FastAPI 的任务执行循环里不断更新TASK_STORE[task_id]["progress"]，把节点名称追加进去。这样做的好处是手机端不用猜 Agent 在干什么，一眼就能看到进度。

4.3 并发和队列：好几个 Agent 同时跑也不慌

手机遥控的第一反应通常是“一口气发好几个任务”，谁都不愿意一个个排队等。但 Agent 任务往往要调模型 API、要请求外部服务，同时开太多并发不仅会拖垮服务端，还可能把第三方 API 的配额打爆。

我给系统加了一个简单的并发控制层，核心是一个信号量，限制同时执行的任务数量：

# task_runner.py import asyncio MAX_CONCURRENT = 3 semaphore = asyncio.Semaphore(MAX_CONCURRENT) async def run_agent_with_limit(task_id, task): async with semaphore: # 实际运行 LangGraph 工作流 app_workflow.invoke(task.params) update_task_store(task_id, "completed")

设置并发上限以后，手机端发 10 个任务，系统会自动把超过上限的任务排在队列里。一开始我不知道自己需要多少并发，实测下来 3 个左右最稳，模型 API 的响应速度和服务端资源都能扛住。等后面真的需要更多并发，再调大信号量和增加 Worker 数量即可。

手机端轮询也要考虑频率问题。我默认设置为 2 秒一次，一个页面同时显示 5 个任务时，请求量是 2.5 QPS，对 FastAPI 来说毫无压力。为了防止用户手滑疯狂点刷新，我在前端做了节流，同一页面最短刷新间隔设为 2 秒。

4.4 远程访问配置：手机在外面也能连上服务

服务端代码搞定以后，剩下的问题是怎么让手机在外面通过公网访问。我直接在云主机上做三层保障：

• 把 FastAPI 服务绑定到本机的127.0.0.1:8080，不直接暴露公网端口。
• 用 Nginx 做反代，对外监听 443 端口，配置 HTTPS 证书。
• 云主机安全组只放行 443 端口，封掉 22 以外的所有端口。

HTTPS 是必须的，因为手机在外面连的是公共 Wi-Fi，如果走明文 HTTP，登录的 Token 很容易被抓包拿到。配置完 HTTPS 和反代以后，手机浏览器直接访问https://your-domain.com就能打开管理页面。

管理页面我也做得特别简朴，就是 HTML + JavaScript，调几个接口展示任务列表、状态和结果。没有用重前端框架，因为这种控制页面的核心就是“直来直去”，不需要复杂交互。为了手机上用得舒服，我把按钮尺寸调大了一些，页面宽度限制在移动端友好的范围内。

5. 踩坑实录：这些问题我折腾了很久

5.1 任务卡死：Agent 掉进循环里出不来

第一次实测的时候，我给 Agent 配了“搜索资料”的工具，让它反复搜索并总结。结果手机端一直显示“执行中”，任务跑了 20 多分钟都没结束。排查以后发现是 Agent 内部陷入了一个“搜索 -> 结果不满足 -> 再搜索”的死循环。

解决方案是给 LangGraph 工作流加了最大步数限制和节点超时控制。LangGraph 支持在编译时设置递归深度，或者给关键节点加timeout，超过时间就自动走失败分支。

这个坑给我的启发是：手机远控场景下，Agent 的“失控时间”会被拉长。电脑前发现卡死还能马上手动终止，手机上发现异常时往往人不在电脑旁，所以必须靠服务端机制兜底。现在我的每个 Agent 都有硬性步数上限，跑超了就自动终止并返回失败原因。

5.2 手机端掉线导致结果丢失

用的人都知道，手机锁屏、切换 App、网络波动都是常态。最开始我的手机端是直接等待同步结果，也就是发完任务后一直保持页面打开。一旦锁屏切走，回来发现结果已经出来了，但页面刷新后状态不同步。

后来我在服务端做了任务结果持久化，即使手机端断线，重连后也能通过task_id拉取到完整结果。持久化用的就是 SQLite，存任务 ID、状态、结果 JSON，查询快、单机零依赖，没必要为了这个场景上重型数据库。

5.3 并发扎堆：手机发了一堆任务，模型 API 被限流

有一次我把 8 个任务一次性推给系统，信号量上限是 3，结果队列里堆满任务，一部份还触发了模型 API 的限流返回 429。排查后发现不是队列的问题，而是我在 Agent 里调用模型 API 时没做重试和退避。

后来给所有外部 API 调用加了一层统一的重试逻辑：遇到 429 或 5xx 错误，等待 1 秒、2 秒、4 秒指数退避重试，最多重试 3 次。加上信号量限流以后，基本没有再出现过 API 被限流的情况。

5.4 端口暴露的安全隐患

一开始为了图省事，我让 FastAPI 直接监听公网端口，结果一晚上收到几百次扫描请求。虽然 Agent 本身有权限控制，但这种裸奔式暴露本身就是让攻击者免费试错。

改掉以后，我确立了“公网只留 HTTPS 443”的原则。服务端代码变化不多，但安全等级完全不一样。

6. 常见问题速查表

自己做这套系统最大的体会是：手机远程操控 AI Agent 的难点，不在于怎么在手机上写代码，而在于把服务端架构设计得足够“耐折腾”。手机只是入口，真正决定体验的是服务端的稳定性、安全性和任务可靠性。把这三件事做好，手机遥控所有 Agent 就是水到渠成的事。

最后再分享一个小技巧：我在手机主屏幕给管理页面建了个快捷方式，打开就像原生 App 一样，完全无浏览器冗余元素。配合桌面小组件显示任务列表，随时扫一眼就知道 Agent 们的状态，体验比坐在电脑前守着还舒服。