★ FEATURED ARTICLE 去年开始聊 AI Agent 的人突然多了起来但我身边大部分人的真实状态是这样的电脑上跑着一个 LangGraph 的本地智能体服务器上挂着一个定时发日报的 Agent手机上还用着一个在线平台的 Bot每个之间互相不认识想远程点一下某个任务要么开电脑要么让队友帮忙完全没有“智能”的感觉。后来我动手做了一个非常轻量的东西把所有 Agent 的“手”收编到一个控制层里手机只需要发一条消息就知道任务在哪个 Agent 上跑、跑得怎么样、结果返回到哪里。整套东西没有用重型的调度平台就是 FastAPI 消息约定 状态同步。这篇文章就把我的设计思路、核心代码、部署细节和踩过的坑完整写出来目标是让同样手里攒了三五个 Agent 的读者也能用一个手机遥控起来。1. 为什么我非要用手机遥控所有 AI Agent1.1 散落各处的 Agent 才是真痛点先说一个我自己的真实场景。我的 Agent 大致分成三类一类是跑在本地笔记本上的研究助手用 LangChain 写了个带工具调用的 RAG 机器人一类是丢在云服务器上的定时任务 Agent每天早上抓取几个信息源然后汇总成简报还有一类是在线平台配置的客服问答机器人处理固定问题。这三类 Agent 的触发方式完全不一样。本地那个要开电脑跑 uvicorn云端那个要 SSH 上去看 log在线平台那个要打开网页后台点测试。有一天我出差客户临时问一个行业数据我明明提前把数据查询 Agent 写好了但就是够不着因为所有入口都绑在固定设备上。那一刻我意识到Agent 数量再多如果控制入口是散的本质上等于没有 Agent。1.2 手机不是算力是控制面把手机当成 Agent 的“遥控器”不是让手机去跑大模型。手机那点算力跑推理不太现实而且真正的 Agent 往往需要调用数据库、文件系统、外部 API这些都放在服务器上更合适。手机适合做的只有一件事把用户的意图转换成一条指令发给正确的 Agent然后接收结果。所以我的核心思路是控制面与执行面分离。执行面是各个 Agent 自己干活的服务器和进程控制面是一层很薄的指令中转服务。手机不再直接连 Agent而是连控制面控制面通过统一的协议把任务分发下去再把结果收回来。这样做的好处是以后我再加一个用 Rust 写的 Agent或者换成 Spring AI Agent手机端完全不用改只需要让新 Agent 按约定接入控制面就行。1.3 一个明确的目标我的验收标准很简单在 4G 网络下手机浏览器打开一个极简控制页点一个按钮远端的 Agent 开始执行页面能实时看到“正在运行”的状态跑完后能显示返回结果。整个过程不超过三秒的延迟而且失败时有明确的错误反馈。这篇文章后面所有的设计和代码都是围绕这个目标来的。2. 整体架构设计一台手机怎么管住所有 Agent2.1 控制层到底该怎么设计我把控制层设计成一个独立的 Web 服务不跟任何具体 Agent 的业务逻辑耦合。每个 Agent 接入之前先注册自己告诉控制层三件事这个 Agent 叫什么名字对应什么能力用什么方式触发HTTP 接口调用、本地命令执行、还是平台 Webhook预计执行多久是否需要异步执行。控制层维护一张注册表把 Agent 的标识符、触发地址、执行方式、状态记录都存进去。手机端请求过来后控制层查表、鉴权、入队、转发表单最后把 Agent 的结果返回。我没有引入消息队列因为个人场景的并发量很低控制层到 Agent 之间直接用 HTTP 同步调用就够。只有当某个 Agent 要跑很长时间比如超过 30 秒时才需要把请求改成异步任务模式先返回一个任务 ID再由前端轮询结果。这里有一个设计取舍要说清楚很多人一上来就上 Celery 或者 Redis Stream但个人项目里 Agent 的数量可能只有几个到十几个请求压力极小。引入消息队列意味着多一个组件要维护多一种故障源手机端调试时还要处理复杂的消费进度。所以我坚持“能用同步解决就不用异步真需要异步再逐层加”这是个人基础设施项目最重要的原则。2.2 指令协议用最简单的 JSON手机和控制层之间、控制层和 Agent 之间我都用 JSON 作为指令协议。一条标准的控制指令长这样{ agent_id: research-assistant, action: run, params: { prompt: 查询2025年AI Agent主流架构相关资料, timeout: 60 }, request_id: a1b2c3d4 }request_id是整个协议里最重要的字段。它是全局唯一的一次请求标识用 UUID 生成作用有两个一是幂等控制层看到相同request_id时不会重复触发 Agent二是追踪日志里全程带着这个 ID出了问题能一条线查到底。2.3 Agent 状态用轮询比 WebSocket 靠谱很多教程会推荐用 WebSocket 做实时状态推送手机端确实能即时收到更新。但我实际用下来发现个人场景里 WebSocket 的维护成本太高断线重连、心跳包、浏览器兼容性都是坑。我更推荐朴素的做法Agent 执行状态写入控制层内存手机端每两秒轮询一次。每次轮询返回一个状态对象{ request_id: a1b2c3d4, agent_id: research-assistant, status: running, message: 正在调用外部检索工具, result: null, updated_at: 2025-01-06T12:30:45Z }轮询的代价是手机端要多写一个 setInterval换来的是整个系统简单可靠。在高并发生产系统里我可能会换方案但个人遥控场景里稳定压倒一切。2.4 并发冲突怎么处理手机只有一个用户但同一个 Agent 可能在同时执行多个任务。比如我让研究助手查资料又让同一个 Agent 生成摘要这两个请求先后进来如果不做控制Agent 内部状态会产生竞态。我的方案是给每个 Agent 加一个执行锁。控制层在同一个 Agent 的请求进来时先检查锁如果 Agent 正在忙新请求不会塞进去而是直接返回 409 状态码提示“该 Agent 正在执行任务”。这个选择很多人会觉得太粗暴为什么不排队呢因为个人场景里排队造成的时间不确定性比直接拒绝要难受手机端看到“忙”可以立刻决定换一个 Agent 或者晚点再试而不是傻等一个未知队列。3. 核心代码实现FastAPI LangGraph 落地3.1 依赖和项目结构我选择了 FastAPI 作为控制层框架理由很实在异步支持好类型校验方便自带接口文档。Agent 本体我用 LangGraph 写了一个带工具调用的 Demo让它能真的“干活”而不是只会回复一句话。项目结构如下agent-controller/ ├── main.py # FastAPI入口 ├── registry.py # Agent注册表 ├── protocol.py # 请求响应模型 ├── agents/ │ ├── research.py # 研究助手 AgentLangGraph实现 │ └── scheduler.py # 定时任务 Agent简单示例 ├── static/ │ └── index.html # 手机端控制页 └── requirements.txt核心依赖就这几个fastapi uvicorn[standard] pydantic langgraph langchain-openai httpx3.2 Agent 统一的执行接口为了让控制层不关心 Agent 的具体实现我给每个 Agent 定义了一个统一的执行函数签名输入一个字典输出一个字典。这是最简单也最难做到的一层抽象。先看注册表实现# registry.py from typing import Callable, Dict, Any agent_registry: Dict[str, Dict[str, Any]] {} def register_agent( agent_id: str, name: str, description: str, executor: Callable[[Dict[str, Any]], Dict[str, Any]], max_timeout: int 30 ): agent_registry[agent_id] { agent_id: agent_id, name: name, description: description, executor: executor, max_timeout: max_timeout, status: idle } def get_agent(agent_id: str): return agent_registry.get(agent_id)这里executor是一个可调用对象它接收 params 字典返回结果字典。控制层完全不关心 executor 内部是 LangGraph、LangChain 还是纯函数——这叫“面向接口编程”后面的扩展性全靠这层抽象。3.3 控制层路由实现控制层对外暴露三个核心接口POST /agent/run提交任务GET /agent/status/{request_id}查询状态GET /agent/list查看所有 Agent任务提交接口是核心我贴出关键实现# main.py from fastapi import FastAPI, HTTPException, Request import uuid from datetime import datetime from registry import agent_registry, get_agent from protocol import RunRequest, StatusResponse app FastAPI(titleAgent Controller) # 用字典保存任务状态真实生产环境可以换 Redis task_store: dict[str, dict] {} agent_locks: dict[str, bool] {} app.post(/agent/run) async def run_agent(req: RunRequest, request: Request): agent get_agent(req.agent_id) if not agent: raise HTTPException(404, fAgent {req.agent_id} 不存在请先注册) # 简单鉴权头信息必须包含预设 token auth_token request.headers.get(X-Agent-Token, ) if auth_token ! your-secret-token: raise HTTPException(401, 无效的访问令牌) # 检查并发锁 if agent_locks.get(req.agent_id, False): raise HTTPException(409, fAgent {req.agent_id} 正在忙请稍后再试) request_id str(uuid.uuid4()) task_store[request_id] { request_id: request_id, agent_id: req.agent_id, status: pending, message: 任务已创建等待执行, result: None, updated_at: datetime.utcnow().isoformat() } # 先设置锁然后异步执行 agent_locks[req.agent_id] True import asyncio asyncio.create_task(_execute_agent(req.agent_id, request_id, req.params)) return {request_id: request_id, status: accepted} async def _execute_agent(agent_id: str, request_id: str, params: dict): agent get_agent(agent_id) task task_store[request_id] try: task[status] running task[message] Agent 已开始执行 task[updated_at] datetime.utcnow().isoformat() # 这里用 asyncio.to_thread 把同步阻塞的 Agent 变成异步任务 result await asyncio.to_thread(agent[executor], params) task[status] success task[message] Agent 执行完成 task[result] result except Exception as e: task[status] failed task[message] str(e) task[result] None finally: agent_locks[agent_id] False task[updated_at] datetime.utcnow().isoformat()几点说明asyncio.create_task让接口立刻返回手机端不用干等。asyncio.to_thread把同步的 Agent 执行放进线程池避免阻塞事件循环。任务状态全部写在内存字典里重启服务会丢失但这套个人场景能够接受如果要求持久化只需要把 task_store 换成 SQLite 或者 Redis 的一个数据结构。状态查询接口更简单app.get(/agent/status/{request_id}) async def get_status(request_id: str): task task_store.get(request_id) if not task: raise HTTPException(404, 任务 ID 不存在) return task3.4 写一个真正干活的 LangGraph Agent控制层写好后重点来了Agent 本体。我用 LangGraph 实现一个研究助手它至少要有规划、调用工具、生成回复三个环节。完整代码较长这里放出核心的图结构# agents/research.py from langgraph.graph import StateGraph, END from typing import TypedDict, List from langchain_openai import ChatOpenAI class ResearchState(TypedDict): prompt: str search_results: str final_answer: str def search_node(state: ResearchState): # 模拟一次外部检索工具的调用 # 实际场景里可以替换成 Bing Search API、本地知识库或数据库查询 query state[prompt] mock_results f针对『{query}』的检索结果AI Agent 主流架构包括 Plan-and-Execute、ReAct、Multi-Agent 协作等... return {search_results: mock_results} def generate_node(state: ResearchState): llm ChatOpenAI(modelgpt-4o-mini, temperature0.3) prompt f基于以下检索资料用中文回答用户问题\n{state[search_results]}\n问题{state[prompt]} resp llm.invoke(prompt) return {final_answer: resp.content} def build_research_graph(): graph StateGraph(ResearchState) graph.add_node(search, search_node) graph.add_node(generate, generate_node) graph.add_edge(search, generate) graph.add_edge(generate, END) graph.set_entry_point(search) return graph.compile()# 注册到控制层 from registry import register_agent async def research_executor(params: dict): graph build_research_graph() result await graph.ainvoke({ prompt: params.get(prompt, 请介绍AI Agent), search_results: , final_answer: }) return {answer: result[final_answer], raw: result} register_agent( agent_idresearch-assistant, name研究助手, description联网检索并总结资料, executorresearch_executor, max_timeout120 )executor是异步函数因为在 LangGraph 里ainvoke是异步的而控制层asyncio.to_thread会把异步函数当作可等待对象处理所以能兼容。如果 Agent 是同步的to_thread也能兜底。3.5 定时任务型 Agent 怎么接入还有一种很常见的 Agent 是定时触发的比如每天早上 8 点抓取信息源生成简报。这种 Agent 不需要手机直接触发但是手机需要知道它上次执行的结果、有没有失败。我用了最原始的方案定时 Agent 本身是一个独立脚本由系统自带的 crontab 调用执行结束后把结果写入一个 JSON 文件。控制层加一个接口直接读取这个文件返回给手机。这比在控制层内部做调度器简单十倍而且即使控制层重启定时任务也不受影响。app.get(/agent/report) async def get_report(): try: with open(/data/report.json, r, encodingutf-8) as f: return json.load(f) except FileNotFoundError: return {message: 今天还没有生成报告}这个设计告诉我一个经验不要把所有逻辑都塞进一个框架里。定时任务用 crontab跨设备通知用手机轮询控制层只做它最擅长的事——接收指令和转发请求。系统总复杂度会大幅下降。4. 实操部署与手机联动全记录4.1 部署到云服务器并配置 HTTPS控制层是标准的 Python Web 服务部署没有任何特殊之处。我把它跑在一台 2 核 2G 的云服务器上操作系统是 Debian用 systemd 管理进程。一个关键点是手机在外网访问控制层必须通过 HTTPS 暴露而且必须加鉴权。否则你的控制层就裸奔在公网上任何人都能调用你的 Agent轻则浪费你的令牌额度重则触发危险工具。我的做法是用 Caddy 作为反向代理它自动申请和续期证书配置只用几行agent.example.com { reverse_proxy 127.0.0.1:8000 }然后云服务商的安全组只放行 80 和 443 端口8000 端口不对外开放。同时 FastAPI 这个应用本身也要校验X-Agent-Token这是第二道防线。两层防护缺一不可。4.2 手机端控制页做成 PWA 风格手机端我一开始想过开发原生 App后来放弃了。个人项目开发原生 App 成本太高而且控制页面的需求就是一张表单加几个按钮浏览器完全够用。我在static/index.html里写了一个极简页面包含三部分Agent 列表显示所有已注册的 Agent名称、状态、最大超时时间。任务提交区输入 prompt 和 agent_id点击执行。任务状态区每两秒轮询当前 request_id 对应的状态显示消息和结果。核心代码是一个轮询函数async function checkStatus(requestId) { const resp await fetch(/agent/status/${requestId}); const data await resp.json(); document.getElementById(status).innerText data.status : data.message; if (data.status success || data.status failed) { clearInterval(timer); if (data.result) { document.getElementById(result).innerText JSON.stringify(data.result, null, 2); } } }手机浏览器访问https://agent.example.com把页面添加到主屏幕体验已经接近原生 App。我实测在 4G 网络下点一下按钮到任务状态变成 running大约 1.5 秒完全在可用范围内。4.3 用手机快捷指令实现“一句话触发”更进一步我把控制层封装成了 iPhone 快捷指令。用快捷指令的“获取 URL 内容”步骤发送 POST 请求输入框里填 prompt 参数。这样锁屏状态下喊一句 Siri就能完成一次 Agent 调用。快捷指令的请求格式是POST https://agent.example.com/agent/run Headers: X-Agent-Token: your-secret-token Content-Type: application/json Body: {agent_id: research-assistant, action: run, params: {prompt: 刚才输入的文本}}快捷指令返回的是 JSON 里的 request_id然后再用一段轮询脚本解析最终结果。这个体验非常接近“赛博助理”的感觉而且全程不用打开任何 App。5. 常见问题与排查避坑速查5.1 问题速查表现象可能原因排查方法提交任务后 409 错误Agent 正在执行锁未释放查看状态接口确认是否 running等几秒重试状态一直是 pending异步任务卡在事件循环创建之前确认 FastAPI 启动时没有禁用异步重启服务Agent 返回 500内部函数抛异常查看 uvicorn 日志找 request_id 对应的 traceback手机访问页面打不开Caddy 证书未生效或防火墙未放行 443检查 Caddy 日志和安全组规则Agent 执行成功但手机看不到结果轮询停止了或者返回结果体积过大确认前端轮询还在跑控制层限制结果最大 1MB同一个 Agent 任务频繁互斥Agent 执行时间过长检查 Agent 内部是否有死循环或外部 API 超时5.2 那些文档里不会写的坑第一个坑是asyncio.to_thread和 LangGraph 兼容性问题。我最初在_execute_agent里直接await agent[executor](params)但遇到某些同步 Agent 会阻塞事件循环。后来统一改成asyncio.to_thread包装再在 Agent 内部自己决定是否异步问题就消失了。原则是控制层永远不在事件循环里直接跑 Agent 的阻塞逻辑。第二个坑是 request_id 幂等。早期没有加幂等时手机端在网络抖动后重试一次同一个任务被触发了两次数据写重了。后来在控制层检查相同 request_id 是否已经存在如果存在就直接返回旧结果彻底解决。第三个坑是超时设置。Agent 调外部接口时HTTP 客户端默认可能无限等待一旦外部 API 卡住整个 Agent 锁死后续所有请求都被 409 拒掉。我的解决办法是给每个 Agent 设置max_timeout外层用asyncio.wait_for包一层超时直接标记失败并释放锁。5.3 还能再往上叠什么这套东西跑通之后我意识到它是一个可以不断生长的基础设施。后续有几个方向值得做把 task_store 从内存换成 Redis 实现多设备状态同步给控制层加一个简单的 token 管理页面不用手写进代码在手机端增加历史任务列表按天查看所有 Agent 的执行记录。此外对于想接入更多 Agent 的读者我的建议是不要一开始就追求大而全的编排。先把一个 Agent 接进来跑通全链路再逐步扩展。每加一个 Agent只需要注册一个函数和一个元信息控制层代码完全不用动这个收益会越来越明显。按我个人经验这套“控制面 执行面分离”的思路价值不在于技术有多深而在于它把散落的 AI 能力收拢到了一个入口。手机从此不再是算力中心而是真正的指挥中心。以后不管是用 LangGraph 还是别的框架只要遵循同一个协议新 Agent 接入都是几分钟的事。
阅读完成 · 觉得有帮助？
