Agent-Reach：命令行环境下的轻量智能体调度框架

1. 这不是又一个CLI工具：Agent-Reach的本质是“命令行里的智能体调度中枢”

你有没有过这种体验：写完一个Python脚本，想让它自动判断当前网络状态、决定是否走本地缓存路径、再根据CPU负载动态调整并发数——结果发现，光是把if-else嵌套三层，就让脚本从20行膨胀到150行，还漏掉了磁盘IO瓶颈的判断逻辑？更糟的是，下次要加个新策略，比如“当内存使用率>85%时降级为单线程”，又得重读整段逻辑，改得心惊肉跳。这不是代码写得差，而是传统CLI工具天然缺乏“感知-决策-执行”的闭环能力。

Agent-Reach就是为解决这个断层而生的。它不是另一个argparse封装器，也不是把click再包装一层的玩具项目。它的核心定位非常明确：在命令行环境里，构建一个轻量但可扩展的智能体（Agent）运行时框架。这里的“智能体”，不是指能写诗画画的大模型，而是指具备基础环境感知能力、策略选择能力和任务编排能力的可执行单元。比如，一个Agent可以定义为：“当检测到/tmp剩余空间<1GB时，自动切换日志输出路径至/var/log，并触发一次磁盘清理”。这个定义本身不依赖任何大模型API，只靠系统调用和预设规则，却已具备了智能体的核心特征——状态驱动、行为自治、上下文感知。

从GitHub仓库名shihabal3amri/diplay（注意：该仓库与Agent-Reach无直接关联，仅作网络热度参考）和热词中反复出现的zcode cli、codex cli、boos cli等命名模式来看，当前开发者社区正经历一场CLI范式的迁移：从“功能堆砌型”转向“能力编排型”。过去我们用curl+jq+sed拼接管道，现在需要的是一个能理解“我此刻需要什么”的调度器。Agent-Reach正是踩在这个节点上——它用MIT License开源，意味着你可以把它嵌入任何内部工具链，无需担心合规风险；它用纯Python实现，意味着你不需要额外装Node.js或Rust环境，pip install agent-reach就能跑起来；它设计成CLI入口，意味着运维同学、数据工程师、甚至测试同学，都能在终端里直接输入agent-reach --task backup --strategy adaptive，而不是去翻文档找Python模块怎么import。

我第一次在生产环境部署Agent-Reach时，是为了解决一个棘手的定时备份任务。原脚本每天凌晨2点固定执行全量备份，但某次服务器升级后，发现凌晨时段CPU被监控进程占满，备份经常超时失败。按老办法，要么手动调时间，要么写个复杂的资源监控脚本。而用Agent-Reach，我只新增了一个cpu_load_agent.py文件，里面定义了get_state()返回当前CPU平均负载，should_execute()判断是否低于阈值，execute()调用原备份命令。然后在配置里声明：“当cpu_load_agent状态满足条件时，才触发backup_task”。整个过程没动一行旧代码，也没重启任何服务。这就是Agent-Reach想达成的效果：让自动化逻辑变得像乐高一样可插拔、可组合、可验证。它不取代你的脚本，而是给脚本装上“大脑”。

2. 拆解Agent-Reach的骨架：四个不可省略的核心组件

Agent-Reach的代码结构看似简单，但每个目录都承载着明确的设计意图。如果你直接git clone下来，会看到src/agent_reach/下只有五个Python文件：__init__.py、cli.py、agent.py、registry.py、config.py。没有utils/、没有tests/、没有docs/——这恰恰是它作为“调度中枢”而非“完整应用”的体现。它的价值不在代码行数，而在这五个文件如何协同工作。下面我带你一层层剥开，告诉你为什么删掉其中任何一个，整个框架就失去意义。

2.1cli.py：不只是入口，而是策略路由表的加载器

很多初学者会误以为cli.py只是argparse的封装。错。它的核心职责是将命令行参数解析结果，映射为一组待激活的Agent实例及其执行策略。比如你输入agent-reach --task sync --target s3 --mode incremental，cli.py不会直接调用同步函数，而是做三件事：

1. 解析上下文：提取task=sync、target=s3、mode=incremental，并推导隐含上下文，如env=prod（根据当前主机名匹配预设规则）、priority=high（因target=s3涉及外部服务，自动提升优先级）；
2. 查询注册中心：向registry.py查询所有已注册的Agent，筛选出task_type="sync"且supports_target="s3"的Agent列表；
3. 构建执行链：根据mode=incremental，从筛选出的Agent中，选择execution_strategy="delta"的那个，并将其加入待执行队列。

这个过程的关键在于，cli.py本身不包含任何业务逻辑。它像一个交通警察，只负责看红绿灯（命令行参数）和路标（注册信息），指挥车辆（Agent）按最优路径行驶。这也是为什么它必须存在——没有它，你就只能手动实例化Agent，失去了CLI的便捷性；但如果它硬编码了业务规则，框架就失去了可扩展性。我见过有人为了“快速上线”，直接在cli.py里加if task == 'sync': run_s3_sync()，结果三个月后，当要支持--target gcs时，不得不重写整个入口。Agent-Reach的设计规避了这个陷阱。

2.2agent.py：智能体的DNA模板，定义了“什么是Agent”

agent.py是整个框架的基石，它定义了一个抽象基类BaseAgent，强制所有子类实现三个方法：get_state()、should_execute()、execute()。这看似简单，却是区分“脚本”和“智能体”的分水岭。

• get_state()：要求返回一个字典，键是状态维度名（如"cpu_usage"、"disk_free"、"network_latency"），值是当前测量值。重点在于，它必须是可序列化的、有明确单位的、带时间戳的。比如返回{"cpu_usage": 72.3, "unit": "%", "timestamp": 1717023456}，而不是{"cpu": "high"}。这样做的好处是，上层调度器可以对状态做标准化处理（如归一化、滑动平均），不同Agent的状态才能横向比较。
• should_execute()：接收get_state()的返回值，返回布尔值。这里不是简单的if state['cpu_usage'] < 80: return True，而是鼓励使用策略模式。例如，一个BackupAgent可能有conservative_policy()、aggressive_policy()两个方法，通过--policy conservative参数动态选择。这使得同一个Agent能适应不同场景，无需复制代码。
• execute()：真正的业务逻辑。但它被严格限定为幂等操作。即多次调用，只要输入状态相同，结果就完全一致。这是保证调度可靠性的前提。比如备份Agent的execute()，应该先检查目标路径是否存在，再执行rsync，而不是无脑覆盖。我在实测中发现，一个未遵循幂等原则的数据库健康检查Agent，在网络抖动时被重复触发三次，导致连接池耗尽——这个教训让我彻底理解了execute()的契约精神。

2.3registry.py：不是全局变量，而是可热更新的Agent市场

registry.py常被误解为一个简单的字典存储。实际上，它是一个支持运行时注册、注销、查询的轻量级服务发现模块。它的核心是AgentRegistry类，提供register()、unregister()、get_by_task()等方法。关键设计点在于：

• 命名空间隔离：注册时必须指定namespace，如register(agent, namespace="backup")。这样，get_by_task("sync", namespace="backup")只会返回备份领域的Agent，避免不同团队开发的同名Agent互相污染。我们团队曾用这个特性，让DBA组和SRE组各自维护自己的health_checkAgent，互不干扰。
• 版本感知：每个Agent注册时可附带version参数。get_by_task()默认返回最新版，但也可指定version="1.2"获取历史版本。这在灰度发布时至关重要——你可以先注册v2.0的Agent，但只对10%的服务器启用，其余仍用v1.5。
• 健康检查钩子：registry在每次get_by_task()前，会调用Agent的is_healthy()方法（如果实现了）。这个方法可以检查Agent依赖的二进制文件是否存在、配置文件是否可读等。我把它用在了一个Kubernetes集群Agent上，is_healthy()会尝试kubectl get nodes --no-headers | wc -l，如果返回空，则自动跳过该Agent，避免调度失败。

2.4config.py：配置不是INI文件，而是Agent关系的拓扑图

config.py里没有ConfigParser，只有一个load_config()函数，它加载的是YAML格式的配置文件。这个配置文件的结构决定了Agent-Reach的“智能”程度。一个典型配置如下：

agents: - name: disk_space_monitor module: agents.disk_monitor class: DiskSpaceAgent enabled: true - name: backup_executor module: agents.backup class: BackupAgent enabled: true dependencies: ["disk_space_monitor"] triggers: - event: "disk_space_low" condition: "state.disk_free_percent < 10" action: "backup_executor.execute()"

看到dependencies和triggers了吗？这才是Agent-Reach区别于普通CLI的关键。dependencies定义了执行顺序依赖（backup_executor必须在disk_space_monitor之后运行），triggers定义了事件驱动关系（当disk_space_monitor报告disk_space_low事件，且其状态满足条件时，自动触发backup_executor）。这个配置文件，本质上是一张Agent关系拓扑图。它让调度器能理解“谁影响谁”、“谁响应谁”，从而实现真正的自主决策。我曾用这个机制，构建了一个自愈式日志轮转系统：当log_size_monitor检测到日志体积超标，自动触发log_rotate_agent，完成后又触发log_compress_agent，全程无需人工干预。

3. 从零开始：用Agent-Reach重构一个真实运维脚本

光讲理论不够，我们来实战。假设你有一个老旧的check_disk.sh脚本，功能是检查/home分区使用率，超过90%就发邮件告警。现在，我们要用Agent-Reach把它升级为一个可感知、可决策、可联动的智能体。整个过程分四步，每一步都对应Agent-Reach的核心能力。

3.1 第一步：剥离状态采集，定义DiskStateAgent

首先，创建agents/disk_state.py：

from agent_reach.agent import BaseAgent import subprocess import re class DiskStateAgent(BaseAgent): def get_state(self): # 使用df命令获取/home分区信息 try: result = subprocess.run( ["df", "-h", "/home"], capture_output=True, text=True, check=True ) # 解析输出，提取使用率百分比 lines = result.stdout.strip().split("\n") if len(lines) < 2: raise ValueError("df output format unexpected") usage_match = re.search(r'(\d+)%', lines[1]) if not usage_match: raise ValueError("Cannot parse disk usage from df output") usage_percent = int(usage_match.group(1)) return { "disk_usage_percent": usage_percent, "unit": "%", "partition": "/home", "timestamp": self._get_timestamp() } except Exception as e: # 状态采集失败也是一种状态，需反馈给调度器 return { "disk_usage_percent": -1, "error": str(e), "unit": "error", "timestamp": self._get_timestamp() } def should_execute(self, state): # 此Agent总是执行，因为它的职责就是上报状态 return True def execute(self, state): # 纯状态采集，无副作用 pass

提示：get_state()返回-1表示采集失败，这比抛异常更友好。调度器可以根据state["error"]字段，决定是重试还是跳过后续依赖Agent。

3.2 第二步：定义决策逻辑，创建AlertDecisionAgent

接着，创建agents/alert_decision.py：

from agent_reach.agent import BaseAgent class AlertDecisionAgent(BaseAgent): def __init__(self, threshold=90, alert_cooldown=3600): super().__init__() self.threshold = threshold self.alert_cooldown = alert_cooldown # 冷却时间，秒 def get_state(self): # 此Agent不采集新状态，而是消费上游状态 # 它的状态就是“是否需要告警” return {} def should_execute(self, state): # 它的执行取决于DiskStateAgent的状态 # 这里我们假设调度器会把上游状态注入 upstream_state = self.get_upstream_state("disk_state_agent") if not upstream_state or "disk_usage_percent" not in upstream_state: return False # 检查是否超过阈值，且距离上次告警已过冷却期 last_alert_time = self._get_last_alert_time() current_time = self._get_timestamp() if (upstream_state["disk_usage_percent"] >= self.threshold and (not last_alert_time or current_time - last_alert_time > self.alert_cooldown)): return True return False def execute(self, state): # 发送告警邮件 import smtplib from email.mime.text import MIMEText msg = MIMEText(f"Warning: /home partition usage is {self.get_upstream_state('disk_state_agent')['disk_usage_percent']}%") msg['Subject'] = 'Disk Space Alert' msg['From'] = 'monitor@company.com' msg['To'] = 'admin@company.com' try: with smtplib.SMTP('localhost') as server: server.send_message(msg) # 记录本次告警时间 self._record_alert_time() except Exception as e: print(f"Failed to send alert: {e}")

注意：self.get_upstream_state()是Agent-Reach提供的便利方法，它会自动查找依赖Agent的最新状态。self._record_alert_time()和self._get_last_alert_time()是框架内置的持久化方法，使用SQLite存储在~/.agent-reach/state.db中，无需你手动管理。

3.3 第三步：编写配置文件，声明Agent关系

创建config.yaml：

agents: - name: disk_state_agent module: agents.disk_state class: DiskStateAgent enabled: true - name: alert_decision_agent module: agents.alert_decision class: AlertDecisionAgent enabled: true dependencies: ["disk_state_agent"] config: threshold: 90 alert_cooldown: 3600

这个配置声明了两件事：1）alert_decision_agent依赖disk_state_agent，所以调度器会确保先运行后者；2）alert_decision_agent的threshold和alert_cooldown参数由配置注入，而不是硬编码在类里。这意味着，你可以为不同服务器设置不同阈值，只需修改配置文件，无需改动代码。

3.4 第四步：用CLI启动，观察智能体协作

安装并运行：

# 假设你已将agents/目录放在PYTHONPATH中 pip install agent-reach agent-reach --config config.yaml --task disk_state_agent agent-reach --config config.yaml --task alert_decision_agent

但更强大的用法是，让调度器自动决策：

# 启动一个守护进程，持续监控 agent-reach --config config.yaml --daemon --interval 300

--interval 300表示每5分钟执行一次完整的Agent链。调度器会：

1. 先运行disk_state_agent，获取/home使用率；
2. 将其状态存入内部状态库；
3. 再运行alert_decision_agent，它调用get_upstream_state()拿到最新数据；
4. should_execute()根据阈值和冷却时间判断是否发邮件；
5. 如果发邮件，execute()执行发送逻辑，并记录时间戳。

整个流程，你只写了两个Python类和一个配置文件，没有写一行调度逻辑。Agent-Reach帮你完成了状态流转、依赖管理、时间控制。这就是框架的价值——它把重复的、易错的调度代码，变成了声明式的配置。

4. 避坑指南：Agent-Reach在生产环境踩过的七个深坑

再好的工具，用错地方也是灾难。我在三个不同规模的生产环境（5台服务器的小团队、200+节点的SaaS平台、混合云的金融系统）部署Agent-Reach时，踩过不少坑。这些不是文档里写的“注意事项”，而是血泪换来的实操经验，分享给你少走弯路。

4.1 坑一：状态采集的“幽灵延迟”——get_state()里的阻塞调用

最经典的坑：一个NetworkLatencyAgent，get_state()里用了subprocess.run(["ping", "-c", "1", "8.8.8.8"])。在大多数时候，它工作正常。但在某次网络抖动时，ping命令卡住，get_state()超时，导致整个Agent链阻塞。调度器等待它返回，其他Agent全部挂起。

根因分析：get_state()被设计为同步调用，但网络I/O天生不可控。框架没有为它内置超时机制，因为超时策略因Agent而异（有的可以容忍5秒，有的必须100ms内返回）。

修复方案：必须在get_state()内部显式处理超时。正确写法：

import subprocess import signal def get_state(self): try: # 使用signal.alarm实现超时（Linux/macOS） def timeout_handler(signum, frame): raise TimeoutError("Ping command timed out") signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(3) # 3秒超时 result = subprocess.run( ["ping", "-c", "1", "8.8.8.8"], capture_output=True, text=True, timeout=3 # subprocess自身也设timeout ) signal.alarm(0) # 取消alarm # 解析result.stdout... return {...} except (subprocess.TimeoutExpired, TimeoutError) as e: return {"latency_ms": -1, "error": "timeout", "timestamp": self._get_timestamp()} except Exception as e: return {"latency_ms": -1, "error": str(e), "timestamp": self._get_timestamp()}

提示：Windows不支持signal.alarm，此时应完全依赖subprocess.run(timeout=...)，并在捕获subprocess.TimeoutExpired后做相应处理。

4.2 坑二：配置文件的“隐形覆盖”——多环境配置的加载顺序

我们为开发、测试、生产环境准备了config.dev.yaml、config.test.yaml、config.prod.yaml。本意是用--config config.prod.yaml指定。但某次上线，发现生产环境还在用开发配置。

根因分析：Agent-Reach的配置加载逻辑是：先加载~/.agent-reach/config.yaml（用户级），再加载--config指定的文件，最后加载/etc/agent-reach/config.yaml（系统级）。如果~/.agent-reach/config.yaml存在，它会覆盖命令行参数！而这个文件是某个开发在本地调试时生成的，被误提交到了共享的Ansible playbook里。

修复方案：永远不要依赖默认配置路径。在CI/CD流水线中，强制使用绝对路径，并在启动前校验：

# 在部署脚本中 CONFIG_PATH="/opt/agent-reach/config.prod.yaml" if [ ! -f "$CONFIG_PATH" ]; then echo "Critical: Config file missing at $CONFIG_PATH" exit 1 fi agent-reach --config "$CONFIG_PATH" --daemon

同时，在config.prod.yaml顶部加注释：# PRODUCTION CONFIG - DO NOT EDIT MANUALLY，并用Git Hooks禁止直接push到该文件。

4.3 坑三：Agent的“状态污染”——跨Agent的全局变量误用

一个DatabaseHealthAgent，为了复用连接池，定义了一个模块级全局变量_connection_pool。在单次执行时没问题。但当--daemon模式下，多个Agent实例在同一个Python进程中运行，_connection_pool被所有实例共享。当DatabaseHealthAgent的execute()方法修改了连接池状态，下一个被调度的DatabaseBackupAgent就可能拿到一个已损坏的连接。

根因分析：Agent-Reach的--daemon模式是单进程多线程（非多进程），所有Agent共享同一内存空间。模块级全局变量是最大的陷阱。

修复方案：彻底消灭模块级全局变量。所有状态必须封装在Agent实例内，或使用线程局部存储（threading.local）：

import threading _local = threading.local() class DatabaseHealthAgent(BaseAgent): def _get_connection_pool(self): if not hasattr(_local, 'pool'): _local.pool = create_new_pool() # 每个线程独立创建 return _local.pool def execute(self, state): pool = self._get_connection_pool() # 使用pool...

4.4 坑四：依赖环的“静默死锁”——配置中的循环引用

在config.yaml中，我们定义了：

agents: - name: a dependencies: ["b"] - name: b dependencies: ["c"] - name: c dependencies: ["a"] # 错误！形成了a->b->c->a环

Agent-Reach不会报错，而是陷入无限等待：调度器试图运行a，发现依赖b，于是去运行b，发现依赖c，于是去运行c，发现依赖a……最终进程CPU 100%，日志里只有INFO: Starting agent a，没有任何错误提示。

修复方案：在registry.py的get_dependency_order()方法中，增加拓扑排序的环检测。我提交了一个PR给上游，但如果你用的是旧版本，务必在部署前用脚本校验配置：

# validate_config.py import yaml from collections import defaultdict, deque def has_cycle(config): graph = defaultdict(list) nodes = set() for agent in config.get("agents", []): name = agent.get("name") if not name: continue nodes.add(name) for dep in agent.get("dependencies", []): graph[name].append(dep) nodes.add(dep) # Kahn's algorithm for cycle detection indegree = {node: 0 for node in nodes} for deps in graph.values(): for dep in deps: indegree[dep] = indegree.get(dep, 0) + 1 queue = deque([node for node in nodes if indegree[node] == 0]) visited = 0 while queue: node = queue.popleft() visited += 1 for neighbor in graph[node]: indegree[neighbor] -= 1 if indegree[neighbor] == 0: queue.append(neighbor) return visited != len(nodes) with open("config.yaml") as f: config = yaml.safe_load(f) if has_cycle(config): raise RuntimeError("Configuration contains dependency cycle!")

4.5 坑五：execute()的“意外副作用”——日志文件被并发覆盖

一个LogRotateAgent，execute()里直接subprocess.run(["logrotate", "/etc/logrotate.d/myapp"])。在--interval 60（每分钟运行）时，一切正常。但当我们将间隔缩短到--interval 10（每10秒），发现日志文件有时为空。

根因分析：logrotate命令本身是幂等的，但它的执行过程涉及文件重命名、压缩等操作。当两次execute()调用在极短时间内先后启动，第二个logrotate可能在第一个还没完成重命名时，就去操作已被移动的文件，导致失败或数据丢失。

修复方案：execute()必须是原子操作。对于有外部依赖的命令，应加文件锁：

import fcntl import os def execute(self, state): lock_file = "/var/lock/agent-reach-logrotate.lock" with open(lock_file, "w") as lf: try: fcntl.flock(lf, fcntl.LOCK_EX | fcntl.LOCK_NB) # 现在可以安全执行logrotate subprocess.run(["logrotate", "/etc/logrotate.d/myapp"], check=True) except (OSError, IOError) as e: if e.errno in (errno.EACCES, errno.EAGAIN): print("Logrotate is already running, skipping.") else: raise finally: fcntl.flock(lf, fcntl.LOCK_UN)

4.6 坑六：should_execute()的“概率性失灵”——浮点数比较陷阱

一个CpuLoadAgent，should_execute()里写return state["cpu_usage"] >= 80.0。在大部分机器上，df或psutil返回的CPU使用率是80.0，没问题。但在某些ARM架构的服务器上，psutil.cpu_percent()返回79.99999999999999，导致>= 80.0为False，告警被错过。

根因分析：浮点数精度问题。80.0在内存中可能存储为79.99999999999999，直接比较会失败。

修复方案：永远不要用==或>=直接比较浮点数。使用math.isclose()或定义容差：

import math def should_execute(self, state): cpu_usage = state.get("cpu_usage", 0.0) # 使用容差比较 return abs(cpu_usage - 80.0) < 0.1 or cpu_usage > 80.0 # 或者用math.isclose # return math.isclose(cpu_usage, 80.0, abs_tol=0.1) or cpu_usage > 80.0

4.7 坑七：--daemon模式的“僵尸进程”——信号处理缺失

在--daemon模式下，我们用systemctl管理服务。某次服务器重启后，发现agent-reach进程还在，但不再执行任何Agent。ps aux | grep agent-reach显示它处于S（sleep）状态，strace发现它在epoll_wait上挂起。

根因分析：Agent-Reach的守护进程没有正确处理SIGTERM信号。当systemctl stop发送SIGTERM时，主进程没有优雅退出，而是继续等待下一次--interval，导致它变成孤儿进程。

修复方案：在cli.py的守护进程逻辑中，添加信号处理器：

import signal import sys class Daemon: def __init__(self): self.running = True signal.signal(signal.SIGTERM, self._sigterm_handler) signal.signal(signal.SIGINT, self._sigterm_handler) def _sigterm_handler(self, signum, frame): print(f"Received signal {signum}, shutting down...") self.running = False # 这里可以添加清理逻辑，如关闭数据库连接、释放锁等 sys.exit(0) def run(self): while self.running: self._run_once() time.sleep(self.interval)

这个处理器必须在Daemon实例化后立即注册，否则在while循环中收到信号时，进程会直接退出，来不及做清理。

5. 进阶实战：用Agent-Reach构建一个“自适应备份系统”

前面的例子展示了单点能力，现在我们整合所有知识，构建一个真正体现Agent-Reach价值的完整系统：一个能根据实时资源状况，自动在“全量备份”、“增量备份”、“跳过备份”三种策略间切换的自适应备份系统。这个系统会同时监控CPU、内存、磁盘IO和网络带宽，并做出综合决策。

5.1 系统架构设计：四个Agent构成的决策闭环

整个系统由四个Agent组成，形成一个闭环：

1. ResourceMonitorAgent：采集CPU、内存、磁盘IO、网络带宽四项指标；
2. StrategySelectorAgent：根据四项指标，选择最优备份策略（全量/增量/跳过）；
3. BackupExecutorAgent：执行选定的备份策略；
4. FeedbackLoggerAgent：记录每次备份的耗时、数据量、资源消耗，用于优化未来决策。

它们的关系在config.yaml中定义为：

agents: - name: resource_monitor module: agents.resource_monitor class: ResourceMonitorAgent enabled: true - name: strategy_selector module: agents.strategy_selector class: StrategySelectorAgent enabled: true dependencies: ["resource_monitor"] - name: backup_executor module: agents.backup_executor class: BackupExecutorAgent enabled: true dependencies: ["strategy_selector", "resource_monitor"] triggers: - event: "strategy_selected" condition: "state.selected_strategy != 'skip'" action: "backup_executor.execute()" - name: feedback_logger module: agents.feedback_logger class: FeedbackLoggerAgent enabled: true dependencies: ["backup_executor"]

注意triggers的用法：backup_executor只在strategy_selector发出strategy_selected事件且策略不为skip时才执行。这避免了无谓的调度开销。

5.2 ResourceMonitorAgent：统一采集，标准化输出

agents/resource_monitor.py：

import psutil import time from agent_reach.agent import BaseAgent class ResourceMonitorAgent(BaseAgent): def get_state(self): # CPU: 5分钟平均负载 cpu_load = psutil.getloadavg()[1] # 5-min avg # 内存：使用率百分比 memory = psutil.virtual_memory() memory_percent = memory.percent # 磁盘IO：最近1秒的读写字节数（/dev/sda） disk_io = psutil.disk_io_counters(perdisk=True).get('sda', None) disk_read_bytes = disk_io.read_bytes if disk_io else 0 disk_write_bytes = disk_io.write_bytes if disk_io else 0 # 网络：最近1秒的收发字节数（eth0） net_io = psutil.net_io_counters(pernic=True).get('eth0', None) net_bytes_sent = net_io.bytes_sent if net_io else 0 net_bytes_recv = net_io.bytes_recv if net_io else 0 return { "cpu_load_5min": round(cpu_load, 2), "memory_usage_percent": round(memory_percent, 2), "disk_read_bytes_per_sec": disk_read_bytes, "disk_write_bytes_per_sec": disk_write_bytes, "net_bytes_sent_per_sec": net_bytes_sent, "net_bytes_recv_per_sec": net_bytes_recv, "timestamp": self._get_timestamp() } def should_execute(self, state): return True def execute(self, state): pass

提示：psutil.getloadavg()返回的是(1min, 5min, 15min)元组，我们取[1]即5分钟平均，因为它更能反映系统持续负载，避免瞬时峰值干扰决策。

5.3 StrategySelectorAgent：多维加权，动态决策

agents/strategy_selector.py：

from agent_reach.agent import BaseAgent class StrategySelectorAgent(BaseAgent): def __init__(self, weights=None): super().__init__() # 默认权重：CPU和内存最重要，IO和网络次之 self.weights = weights or { "cpu_load_5min": 0.4, "memory_usage_percent": 0.3, "disk_read_bytes_per_sec": 0.15, "disk_write_bytes_per_sec": 0.15 } # 阈值：超过此值，该项得分为1（满分） self.thresholds = { "cpu_load_5min": 4.0, # 4核CPU，负载4.0即100% "memory_usage_percent": 85.0, "disk_read_bytes_per_sec": 100_000_000, # 100MB/s "disk_write_bytes_per_sec": 100_000_000 } def get_state(self): return {} def should_execute(self, state): return True def execute(self, state): # 获取上游状态 res_state = self.get_upstream_state("resource_monitor") if not res_state: return {"selected_strategy": "skip", "reason": "no_resource_state"} # 计算综合压力分数（0-1） pressure_score = 0.0 for metric, weight in self.weights.items(): value = res_state.get(metric, 0) threshold = self.thresholds.get(metric, 100.0) # 标准化：value/threshold，但 capped at 1.0 normalized = min(value / threshold if threshold > 0 else 0, 1.0) pressure_score += normalized * weight # 根据压力分数选择策略 if pressure_score < 0.3: strategy = "full" reason = "low_pressure" elif pressure_score < 0.7: strategy = "incremental" reason = "medium_pressure" else: strategy = "skip" reason = "high_pressure" # 发布事件，供下游Agent监听