1. 边缘计算与 Agent 的碰撞为什么轻量化部署是绕不开的坎边缘计算这个词这几年被聊得很多但很多人对它的理解还停留在“把服务器搬到离用户近的地方”。实际上一个边缘计算节点并不等于一个机房它可能是一台工控机、一块 Jetson 开发板、一个树莓派甚至是一台带 NPU 的国产 ARM 盒子。这些设备的共同特点是算力有限、内存紧张、功耗敏感、网络不稳定。而 Agent 这个东西天生就带着“大模型推理 工具调用 记忆管理 多轮编排”的重负担把它塞进边缘设备里就像让一个习惯了住五星级酒店的人去住胶囊旅馆——不是不能住是得把行李扔掉一大半。我过去一年在几个工业质检和园区安防的项目里反复折腾过 Agent 在边缘侧的落地。踩过的坑包括但不限于模型加载直接把 4GB 内存的盒子干到 OOM、工具调用链路太长导致响应超过 3 秒、记忆模块写满 eMMC 导致设备卡死、多 Agent 协作时消息队列把 CPU 跑满。这些问题的根源都是没有做好轻量化部署。轻量化不是简单地把模型换小一号而是从模型选型、推理框架、Agent 架构、记忆策略、通信机制到资源调度的一整套系统性取舍。这篇文章面向的是已经在做或准备做边缘 Agent 落地的开发者尤其是那些手头设备算力在 1 TOPS 到 20 TOPS 之间、内存 2GB 到 8GB 之间的场景。我会把整个轻量化部署的实践拆开来讲包括为什么这么选、怎么算资源账、每一步具体怎么操作、遇到问题怎么排查。读完之后你应该能拿着一块边缘板子从零搭出一个能跑起来、响应可接受、长期运行不崩的轻量 Agent。2. 整体设计思路把 Agent 拆成能塞进边缘的零件2.1 边缘 Agent 的核心约束与设计原则在动手之前先把约束条件列清楚。边缘设备和云端服务器最大的区别在于资源是“硬上限”不是“弹性伸缩”。我一般会先算三笔账内存账操作系统占 300MB 到 800MB推理框架运行时占 200MB 到 500MB模型权重占大头Agent 的上下文和记忆占剩余部分。如果设备只有 4GB 内存留给模型和 Agent 逻辑的空间可能只有 2GB 左右。算力账以 INT8 量化后的模型为例1 TOPS 算力大约能支撑 1B 参数模型以 5 到 10 token/s 的速度推理。如果要跑到 20 token/s 以上要么模型小于 1B要么算力大于 5 TOPS。存储账eMMC 的写入寿命有限Agent 的记忆模块如果频繁写盘几个月就能把存储写坏。所以记忆必须分层热数据放内存冷数据才落盘且要控制写入频率。基于这三笔账我总结出边缘 Agent 轻量化的四个设计原则模型小型化、推理本地化、记忆分层化、编排精简 化。模型小型化不是一味追求小而是在任务精度可接受的前提下选最小的推理本地化意味着不依赖云端 API所有推理在设备内完成记忆分层化是把短期记忆放内存、长期记忆做摘要后落盘编排精简 化是砍掉不必要的 Agent 循环和工具调用层级。2.2 为什么选择“小模型 规则引擎 轻量 Agent 框架”的组合很多人一上来就想在边缘跑 7B 模型觉得效果才好。实测下来在 8GB 内存的 ARM 设备上跑 7B INT4 模型推理速度大概 2 到 4 token/s一个稍微复杂点的 Agent 任务要等十几秒用户体验直接崩掉。所以我的选择是1B 到 3B 的量化模型做语义理解和生成规则引擎做确定性判断轻量 Agent 框架做编排。这个组合的逻辑是边缘场景里很多任务是确定性的比如“检测到温度超过阈值就触发告警”这种根本不需要大模型规则引擎毫秒级搞定。只有涉及自然语言理解、模糊匹配、多步推理的部分才交给小模型。Agent 框架的作用是把这两者串起来根据任务类型决定走规则还是走模型。框架选型上我没有用 LangChain 这类偏重的框架而是自己写了一个大概 500 行的轻量编排层。原因很简单LangChain 的依赖太多光 Python 包就几十个在边缘设备上装完占几百 MB而且很多抽象层在边缘场景用不上。自己写的编排层只保留三个核心能力任务路由、工具注册、上下文管理。这样整个 Agent 运行时的内存占用能控制在 50MB 以内。2.3 模型选型的量化对比与决策依据模型选型我做过一轮比较系统的测试在 Jetson Orin Nano 8GB 和 RK3588 两款设备上跑了几个候选模型。测试任务是园区场景下的“异常事件描述生成”和“工单分类”评估指标是准确率、推理速度、内存占用。模型参数量量化方式内存占用推理速度任务准确率Qwen2-0.5B0.5BINT8约 600MB25 token/s78%Qwen2-1.5B1.5BINT4约 1.2GB12 token/s86%Phi-3-mini3.8BINT4约 2.4GB6 token/s89%TinyLlama-1.1B1.1BINT4约 900MB15 token/s81%最终我选了 Qwen2-1.5B INT4 作为主力模型原因是它在准确率和速度之间取得了比较好的平衡。0.5B 虽然快但在工单分类这种需要一定语义理解的任务上准确率掉得厉害3.8B 准确率只高了 3 个百分点但速度慢了一半内存也多占一倍。这个选择不是绝对的如果你的场景对准确率要求极高且设备内存有 8GB 以上Phi-3-mini 也可以考虑。注意量化方式的选择很关键。INT8 对精度影响小但内存节省有限INT4 内存节省明显但某些模型会出现明显的精度下降。建议在目标设备上实测不要只看论文数据。3. 核心细节解析从模型加载到 Agent 编排的实操要点3.1 模型量化与推理框架的搭配选择模型量化不是拿个工具跑一遍就完事不同推理框架对量化的支持程度差异很大。我试过三种组合第一种是ONNX Runtime ONNX 量化模型。优点是跨平台好ARM 和 x86 都能跑量化工具链成熟。缺点是对某些算子的支持不完整比如 Qwen2 的 RoPE 旋转位置编码在 ONNX 导出时容易出问题需要手动改图。第二种是llama.cpp GGUF 格式。这是我在边缘设备上用得最多的方案。llama.cpp 对 ARM 的 NEON 指令集优化很好GGUF 格式支持 Q4_K_M 这种混合量化能在精度和体积之间做细粒度权衡。实测 Qwen2-1.5B 转成 Q4_K_M 后模型文件约 1.1GB加载后内存占用约 1.3GB在 RK3588 上能跑到 12 token/s。第三种是厂商专用推理引擎比如瑞芯微的 RKNN、地平线的 BPU SDK。这类引擎对自家芯片优化最好但通用性差模型转换麻烦适合产品化阶段深度绑定硬件的场景。我一般推荐先用 llama.cpp 做原型验证因为它的部署最简单一个二进制文件加一个模型文件就能跑。等验证完再考虑是否迁移到厂商专用引擎做性能压榨。具体操作上把 HuggingFace 模型转 GGUF 的流程是这样的# 克隆 llama.cpp 仓库 git clone https://github.com/ggerganov/llama.cpp cd llama.cpp # 安装 Python 依赖 pip install -r requirements.txt # 转换模型为 GGUF 格式以 Qwen2-1.5B 为例 python convert_hf_to_gguf.py /path/to/Qwen2-1.5B-Instruct \ --outfile qwen2-1.5b-f16.gguf \ --outtype f16 # 量化为 Q4_K_M ./llama-quantize qwen2-1.5b-f16.gguf qwen2-1.5b-q4km.gguf Q4_K_M量化完成后用 llama-server 启动推理服务./llama-server -m qwen2-1.5b-q4km.gguf \ --host 0.0.0.0 --port 8080 \ --ctx-size 2048 \ --threads 4 \ --n-gpu-layers 0这里的参数需要根据设备调整。--ctx-size控制上下文长度边缘场景一般 2048 够用设太大内存会爆。--threads设成 CPU 核心数但不要超过物理核心数超了反而因为调度开销变慢。--n-gpu-layers在有 GPU 或 NPU 的设备上可以设大一点把部分层卸载到加速器上。3.2 Agent 编排层的轻量化实现编排层是整个 Agent 的大脑负责决定“用户输入进来后先干什么、再干什么”。在边缘场景编排层必须极简我把它设计成三个模块任务路由器根据输入的特征决定走规则还是走模型。比如输入里包含明确的设备 ID 和数值直接走规则输入是自然语言描述走模型。路由逻辑用简单的关键词匹配和正则就能实现不需要模型参与。工具注册表把所有可调用的工具注册成统一的接口。每个工具定义名称、描述、参数 schema 和执行函数。工具可以是本地函数比如读传感器、写数据库也可以是 HTTP 调用比如调另一个边缘节点的服务。上下文管理器维护对话历史和任务状态。这里的关键是控制上下文长度不能无限增长。我的做法是保留最近 5 轮对话的完整内容更早的做摘要压缩。编排层的核心循环大概长这样class LightweightAgent: def __init__(self, llm_client, tools, max_context_turns5): self.llm llm_client self.tools tools self.context [] self.max_context_turns max_context_turns def route(self, user_input): # 规则优先检测是否包含确定性指令 if self._match_rule_pattern(user_input): return self._execute_rule(user_input) # 否则走模型推理 return self._execute_llm(user_input) def _execute_llm(self, user_input): # 构建 prompt包含工具描述和上下文 prompt self._build_prompt(user_input) response self.llm.generate(prompt, max_tokens256) # 解析模型输出判断是否需要调用工具 tool_call self._parse_tool_call(response) if tool_call: result self.tools[tool_call[name]](**tool_call[args]) # 把工具结果再喂给模型生成最终回复 final self.llm.generate( self._build_prompt(user_input, tool_resultresult), max_tokens128 ) return final return response def _build_prompt(self, user_input, tool_resultNone): # 精简 prompt只保留必要信息 system 你是一个边缘设备助手简洁回答。 tools_desc \n.join([ f{name}: {t[desc]} for name, t in self.tools.items() ]) ctx self._get_recent_context() parts [system, f可用工具:\n{tools_desc}, ctx, f用户: {user_input}] if tool_result: parts.append(f工具结果: {tool_result}) return \n.join(parts)这个实现只有一百多行但覆盖了边缘 Agent 的核心需求。关键设计点是规则优先、模型兜底、上下文截断、工具结果二次推理。规则优先保证了确定性任务的响应速度模型兜底处理模糊输入上下文截断防止内存无限增长工具结果二次推理让 Agent 能基于工具返回的数据生成自然语言回复。3.3 记忆系统的分层设计与落盘策略Agent 的记忆是边缘部署里最容易出问题的部分。我见过太多项目因为记忆模块把设备写死。核心问题是eMMC 的擦写寿命通常只有 3000 到 5000 次 P/E cycle如果 Agent 每轮对话都写一次盘一天几百次写入几个月就废了。我的分层记忆设计是这样的短期记忆内存保存最近 N 轮对话的完整内容用 Python 的 deque 实现超过长度自动丢弃最旧的。这部分不落盘设备重启就丢但边缘场景通常不需要跨重启保留对话。工作记忆内存 定期落盘保存当前任务的中间状态比如“正在处理的工单 ID”“已经查询过的传感器列表”。这部分每隔 5 分钟或任务完成时落盘一次用 SQLite 存储。长期记忆落盘 摘要保存历史任务的摘要和关键结论。不是保存原始对话而是让模型生成一段 50 字以内的摘要然后存到 SQLite。这样既保留了知识又控制了存储增长。落盘策略上我用了一个写缓冲队列。所有写操作先进入内存队列后台线程每 30 秒批量刷一次盘。这样即使 Agent 每秒产生多次写操作实际落盘频率也被控制在每分钟 2 次以内。SQLite 开启 WAL 模式减少锁竞争。import sqlite3 import threading import queue import time class TieredMemory: def __init__(self, db_path, short_term_size10): self.short_term deque(maxlenshort_term_size) self.working {} self.write_queue queue.Queue() self.db_path db_path self._init_db() self._start_flush_thread() def _init_db(self): conn sqlite3.connect(self.db_path) conn.execute(PRAGMA journal_modeWAL) conn.execute( CREATE TABLE IF NOT EXISTS long_term ( id INTEGER PRIMARY KEY AUTOINCREMENT, summary TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def _start_flush_thread(self): def flush_loop(): while True: batch [] try: while len(batch) 10: batch.append(self.write_queue.get(timeout30)) except queue.Empty: pass if batch: conn sqlite3.connect(self.db_path) conn.executemany( INSERT INTO long_term (summary) VALUES (?), [(item,) for item in batch] ) conn.commit() conn.close() t threading.Thread(targetflush_loop, daemonTrue) t.start() def add_long_term(self, summary): self.write_queue.put(summary)这个实现把写盘频率从“每轮对话”降到“每 30 秒批量”eMMC 寿命问题基本解决。实测在每天 500 轮对话的负载下eMMC 写入量从每天约 50MB 降到每天约 2MB。4. 完整实操过程从零搭建一个边缘 Agent 节点4.1 硬件选型与系统环境准备先说硬件。我目前主力测试设备是 RK3588 开发板8GB 内存6 TOPS NPU和 Jetson Orin Nano8GB 内存40 TOPS。RK3588 性价比高适合量产Jetson 算力强适合原型验证。如果预算有限树莓派 5 8GB 也能跑 0.5B 到 1B 的模型但速度会慢一些。系统环境我统一用 Ubuntu 22.04 ARM64 版本。刷机后先做几件事# 更新系统 sudo apt update sudo apt upgrade -y # 安装基础依赖 sudo apt install -y python3-pip python3-venv cmake build-essential \ libsqlite3-dev sqlite3 curl # 创建虚拟环境 python3 -m venv /opt/agent-env source /opt/agent-env/bin/activate # 安装 Python 依赖 pip install requests flask numpy这里没有装 PyTorch因为边缘推理用 llama.cpp 的 C 后端就够了PyTorch 在 ARM 上装起来又大又慢。Flask 是用来做 Agent 的 HTTP 接口方便其他系统调用。系统调优方面有几个参数必须改# 关闭 swap防止内存不足时频繁换页导致卡死 sudo swapoff -a sudo sed -i /swap/d /etc/fstab # 调整 CPU 调度器为 performance 模式 echo performance | sudo tee /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor # 增加文件描述符限制 echo fs.file-max 65535 | sudo tee -a /etc/sysctl.conf关闭 swap 这个操作在边缘设备上很重要。因为 eMMC 的随机读写速度远低于内存一旦发生 swap推理速度会掉到十分之一以下而且会加速存储磨损。宁可让程序 OOM 被杀掉重启也不要让它 swap。4.2 模型部署与推理服务启动模型文件准备好之后把 llama.cpp 编译成 ARM 版本。编译时要开启 NEON 优化cd llama.cpp mkdir build cd build cmake .. -DCMAKE_BUILD_TYPERelease \ -DLLAMA_NATIVEON \ -DLLAMA_NEONON make -j$(nproc)编译完成后把 llama-server 和模型文件放到 /opt/agent/models 目录。然后创建一个 systemd 服务让推理服务开机自启# /etc/systemd/system/llama-server.service [Unit] DescriptionLlama Inference Server Afternetwork.target [Service] Typesimple Useragent WorkingDirectory/opt/agent ExecStart/opt/agent/llama-server \ -m /opt/agent/models/qwen2-1.5b-q4km.gguf \ --host 127.0.0.1 --port 8080 \ --ctx-size 2048 \ --threads 4 \ --parallel 1 Restartalways RestartSec5 [Install] WantedBymulti-user.target注意--parallel 1这个参数。它控制并发请求数边缘设备上设成 1 就行设大了内存会成倍增长。如果确实需要处理并发更好的做法是在 Agent 层做请求队列而不是让推理服务并行。启动服务并验证sudo systemctl daemon-reload sudo systemctl enable llama-server sudo systemctl start llama-server # 测试推理 curl http://127.0.0.1:8080/completion \ -H Content-Type: application/json \ -d {prompt: 你好, max_tokens: 32}如果返回正常说明推理服务跑起来了。这时候可以看一下内存占用ps aux | grep llama-server # 观察 RSS 列正常应该在 1.2GB 到 1.5GB 之间4.3 Agent 服务封装与接口设计Agent 服务我用 Flask 封装对外提供两个接口一个同步接口用于简单查询一个异步接口用于复杂任务。同步接口直接返回结果异步接口返回任务 ID客户端轮询获取结果。from flask import Flask, request, jsonify import requests import threading import uuid app Flask(__name__) agent LightweightAgent( llm_clientLlamaClient(http://127.0.0.1:8080), toolsregister_default_tools() ) task_store {} app.route(/agent/query, methods[POST]) def sync_query(): data request.json user_input data.get(input, ) if not user_input: return jsonify({error: empty input}), 400 try: result agent.route(user_input) return jsonify({result: result}) except Exception as e: return jsonify({error: str(e)}), 500 app.route(/agent/task, methods[POST]) def async_task(): data request.json task_id str(uuid.uuid4()) task_store[task_id] {status: pending, result: None} def run(): try: result agent.route(data.get(input, )) task_store[task_id] {status: done, result: result} except Exception as e: task_store[task_id] {status: error, result: str(e)} threading.Thread(targetrun, daemonTrue).start() return jsonify({task_id: task_id}) app.route(/agent/task/task_id, methods[GET]) def get_task(task_id): task task_store.get(task_id) if not task: return jsonify({error: task not found}), 404 return jsonify(task) if __name__ __main__: app.run(host0.0.0.0, port5000, threadedTrue)这个服务设计的关键点是同步接口用于快速响应异步接口用于耗时任务。边缘场景里很多 Agent 任务需要多步推理和工具调用耗时可能超过 10 秒同步接口会超时。异步接口把任务丢到后台线程客户端轮询体验更好。接口设计上我建议统一用 JSON 格式输入字段叫input输出字段叫result。这样客户端不需要关心 Agent 内部逻辑只负责发文本和收文本。4.4 资源监控与自动降级机制边缘设备长期运行必须有资源监控和自动降级。我实现了一个简单的监控模块每 10 秒检查一次内存和 CPU 使用率超过阈值就触发降级。import psutil import time class ResourceMonitor: def __init__(self, mem_threshold85, cpu_threshold90): self.mem_threshold mem_threshold self.cpu_threshold cpu_threshold self.degraded False def check(self): mem psutil.virtual_memory().percent cpu psutil.cpu_percent(interval1) if mem self.mem_threshold or cpu self.cpu_threshold: self.degraded True return degraded self.degraded False return normal def get_strategy(self): if self.degraded: # 降级策略缩短上下文、减少 max_tokens、跳过非必要工具 return { max_context_turns: 2, max_tokens: 64, skip_optional_tools: True } return { max_context_turns: 5, max_tokens: 256, skip_optional_tools: False }降级策略的核心是牺牲质量保可用性。内存紧张时把上下文从 5 轮降到 2 轮max_tokens 从 256 降到 64跳过非必要的工具调用。这样虽然回答质量下降但至少不会 OOM 崩溃。监控模块可以集成到 Agent 的编排循环里每次处理请求前检查一次资源状态动态调整策略。实测下来这个机制能在内存使用率达到 90% 时把峰值压回 75% 左右给系统留出缓冲空间。5. 常见问题与排查技巧实录5.1 模型加载失败与内存不足的排查路径模型加载失败是边缘部署最常见的问题表现通常是 llama-server 启动后立即退出或者加载到一半卡死。排查路径我总结成三步第一步看日志。llama-server 启动时会输出模型加载的详细信息包括每一层的加载进度和内存分配情况。如果看到failed to allocate或out of memory基本可以确定是内存不够。第二步算内存账。模型文件大小不等于内存占用。GGUF 格式的模型加载后内存占用大约是文件大小的 1.1 到 1.3 倍因为除了权重还有 KV cache 和运行时开销。比如 1.1GB 的 Q4_K_M 模型实际内存占用约 1.3GB 到 1.5GB。如果设备只有 2GB 内存减去系统占用的 500MB剩 1.5GB刚好卡在边界上很容易失败。第三步降配置。如果确认是内存问题有几个调整方向换更小的模型、用更激进的量化Q4_K_M 换 Q3_K_S、减小 ctx-size、减少 threads。其中减小 ctx-size 效果最明显因为 KV cache 的大小和 ctx-size 成正比。ctx-size 从 2048 降到 1024KV cache 能省一半内存。实操心得在 4GB 内存的设备上我建议模型文件不要超过 1GBctx-size 不要超过 1024threads 不要超过 4。这样留出的余量能保证系统稳定运行。5.2 推理速度慢的优化手段与实测数据推理速度慢的原因通常有三个模型太大、线程数设置不合理、没有启用硬件加速。我做过一组对比测试在 RK3588 上跑 Qwen2-1.5B Q4_K_M不同配置下的速度差异很大配置threadsctx-sizeNPU 加速推理速度默认82048否6 token/s优化线程42048否11 token/s减小上下文41024否13 token/s启用 NPU41024是22 token/s从数据可以看出线程数从 8 降到 4 反而速度翻倍原因是 RK3588 有 4 个大核和 4 个小核设成 8 线程时任务会被调度到小核上拖慢整体速度。ctx-size 减半带来约 18% 的速度提升。启用 NPU 加速效果最明显速度直接翻倍。NPU 加速的配置方法因芯片而异。RK3588 需要用 RKNN 工具链把模型转成 rknn 格式然后在 llama.cpp 的 RK3588 分支里启用 NPU 后端。这个过程比较折腾但性能收益值得投入。另一个容易被忽略的优化点是prompt 长度。Agent 的 prompt 里包含了工具描述、上下文、系统指令很容易超过 500 token。prompt 越长首 token 延迟越大。我的做法是把工具描述压缩到最短系统指令控制在 50 字以内上下文只保留最近 2 轮。这样 prompt 长度能控制在 300 token 以内首 token 延迟从 2 秒降到 0.8 秒。5.3 Agent 循环卡死与工具调用超时的处理Agent 循环卡死是另一个高频问题表现是请求发出去后一直不返回CPU 占用不高但就是不结束。原因通常是模型输出了无法解析的工具调用格式导致编排层陷入死循环。我的处理方案是加两层保护第一层最大循环次数限制。在编排循环里设一个计数器超过 5 次工具调用就强制退出返回当前已有结果。边缘场景的任务通常不需要超过 3 次工具调用5 次是安全上限。第二层工具调用超时。每个工具执行时加超时控制超过 10 秒就中断并返回错误。用 Python 的concurrent.futures实现from concurrent.futures import ThreadPoolExecutor, TimeoutError executor ThreadPoolExecutor(max_workers2) def call_tool_with_timeout(tool_func, args, timeout10): future executor.submit(tool_func, **args) try: return future.result(timeouttimeout) except TimeoutError: future.cancel() return {error: tool timeout}这两层保护加上之后Agent 卡死的概率从每天几次降到几乎为零。即使模型输出了异常格式最多 5 轮循环加 10 秒超时50 秒内一定能返回结果。5.4 常见问题速查表问题现象可能原因排查方法解决方案模型加载失败内存不足看日志中的 allocate 错误换小模型或减小 ctx-size推理速度慢线程数不合理对比不同 threads 设置设为物理大核数Agent 卡死工具调用死循环看日志中工具调用次数加最大循环限制和超时设备重启后 Agent 不启动systemd 服务未启用systemctl statussystemctl enable记忆丢失落盘失败检查 SQLite 文件权限修正文件权限和目录响应时间波动大资源竞争监控 CPU 和内存启用降级策略eMMC 写入量高频繁落盘iostat看写入改批量落盘模型输出乱码量化精度损失对比不同量化级别换 Q5_K_M 或 Q8_0这张表是我在实际项目中遇到问题后整理的基本覆盖了边缘 Agent 部署 80% 的常见故障。建议打印出来贴在工位上出问题时先对照排查。6. 多 Agent 协作在边缘的轻量实现6.1 边缘多 Agent 的通信机制选择单 Agent 跑通之后下一步自然是多 Agent 协作。边缘场景的多 Agent 和云端不一样云端可以用消息队列、gRPC、服务网格边缘设备上这些中间件太重了。我的选择是HTTP 共享内存的混合方案。同一台设备上的多个 Agent 通过共享内存通信用 Python 的multiprocessing.shared_memory实现延迟在微秒级。跨设备的 Agent 通过 HTTP 通信用 Flask 暴露接口延迟在毫秒级。这样既保证了同设备协作的效率又保持了跨设备协作的简单性。通信协议上我定义了一个极简的消息格式{ from: agent-01, to: agent-02, type: task, payload: { action: classify, data: 设备温度异常 }, timestamp: 1700000000 }消息类型只有三种task请求执行任务、result返回结果、heartbeat心跳。没有复杂的路由和订阅机制就是点对点直发。这样实现简单调试方便在边缘场景够用。6.2 任务分发与结果聚合的轻量编排多 Agent 协作的核心是任务分发和结果聚合。我的做法是设一个协调者 Agent它不执行具体任务只负责拆分任务和汇总结果。其他 Agent 是工作者只执行具体任务。协调者的逻辑很简单收到任务后根据任务类型查路由表决定发给哪些工作者。然后并行发送请求等所有结果返回后聚合。聚合逻辑可以是简单的拼接也可以是让模型做一次总结。class CoordinatorAgent: def __init__(self, workers): self.workers workers # {classifier: http://..., summarizer: http://...} def dispatch(self, task): task_type task.get(type) if task_type classify_and_summarize: # 先分类再总结 classify_result self._call_worker(classifier, task) summarize_task {**task, classify_result: classify_result} summarize_result self._call_worker(summarizer, summarize_task) return {classify: classify_result, summary: summarize_result} elif task_type parallel_check: # 并行检查多个传感器 results {} for name, url in self.workers.items(): if name.startswith(sensor): results[name] self._call_worker(name, task) return results def _call_worker(self, name, task): url self.workers[name] resp requests.post(f{url}/agent/query, json{input: task}, timeout15) return resp.json()这个协调者只有几十行代码但能支持串行和并行两种编排模式。实测在 3 个 Agent 协作的场景下端到端延迟在 3 到 5 秒之间对于边缘场景可以接受。6.3 多 Agent 场景下的资源竞争与隔离多 Agent 跑在同一台设备上资源竞争是必须解决的问题。我的方案是进程隔离 资源配额。每个 Agent 跑在独立的进程里用 systemd 的MemoryMax和CPUQuota限制资源。# /etc/systemd/system/agent-worker.service [Unit] DescriptionAgent Worker %i Afterllama-server.service [Service] Typesimple Useragent ExecStart/opt/agent-env/bin/python /opt/agent/worker.py --name %i MemoryMax1G CPUQuota150% Restartalways [Install] WantedBymulti-user.targetMemoryMax1G限制每个 Agent 最多用 1GB 内存超了会被 OOM killer 杀掉然后自动重启。CPUQuota150%限制最多用 1.5 个核心的算力。这样即使某个 Agent 出问题也不会影响其他 Agent 和推理服务。推理服务本身是共享的多个 Agent 通过 HTTP 调用同一个 llama-server。这里要注意并发控制llama-server 的--parallel参数设成 1Agent 层用队列串行化请求。如果确实需要并发可以在设备上起两个 llama-server 实例分别监听不同端口但内存占用会翻倍。注意多 Agent 场景下内存是最容易出问题的资源。建议在部署前用stress-ng做压力测试模拟所有 Agent 同时满载的情况观察内存峰值。如果峰值超过设备内存的 80%就需要减少 Agent 数量或降低单个 Agent 的资源配额。7. 实际项目中的经验体会我在一个园区安防项目里部署了这套方案设备是 4 台 RK3588 盒子每台跑一个 Agent 节点分别负责视频分析、门禁事件、环境监测和工单处理。四个 Agent 通过 HTTP 协作协调者跑在其中一个节点上。运行了半年多整体稳定平均响应时间 2.8 秒内存峰值控制在 6.2GB 以内没有出现过 OOM 或存储写坏的情况。踩过最大的坑是早期没有做记忆分层Agent 每轮对话都写 SQLite结果两个月后一台设备的 eMMC 出现坏块。后来改成批量落盘加摘要压缩写入量降了 95%问题解决。另一个坑是模型量化级别选得太激进用了 Q3_K_S结果工单分类准确率从 86% 掉到 72%后来换回 Q4_K_M 才恢复。如果让我给准备做边缘 Agent 的人一条建议那就是先算资源账再选模型最后写代码。很多人反过来先写代码再发现跑不起来返工成本很高。资源账算清楚了后面的选型和实现都是水到渠成的事。这个方案后续还可以扩展的方向包括用 NPU 做模型推理的进一步加速、引入更细粒度的资源监控和动态调度、支持 Agent 的热更新而不中断服务。这些我还在陆续折腾有新的进展再分享。