用 FROST 五维元模型构建可治理的多 Agent 系统:从零到一的代码教程
# 用 FROST 五维元模型构建可治理的多 Agent 系统:从零到一的代码教程
**作者** :FROST Team
**日期** :2026-07-09
**主题** :代码教程 | 周四轮换
**项目** :FROST + FROST-SOP
## 前言
2026 年,Agent 框架百花齐放——LangChain、CrewAI、AutoGen 各有各的好。但它们都有一个共同的盲区:**治理能力** 。
当你的 Agent 系统从 1 个变成 10 个,从跑 Demo 变成跑生产,你会发现:
* 谁有权做什么?没有答案
* 这个决策是谁做的?无法追溯
* Agent 越权了怎么办?事后才能发现
FROST(Fractal Runtime of Orchestrated Skills & Tasks)的 **五维元模型** 就是为解决这些问题而设计的。
今天这篇教程,我们用代码从零构建一个完整的多 Agent 治理系统。
> **FROST** (教学框架)提供理论基础 → Gitee 仓库
> **FROST-SOP** (工程平台)提供工程落地 → Gitee 仓库
## 一、五维元模型是什么?
FROST V4.0 引入了五个核心维度,每个维度解决 Agent 治理的一个关键问题:
维度 | 模块 | 解决的问题 | 类比
---|---|---|---
**武器** | Armory | Agent 有哪些能力? | 武器库清单
**任务** | TaskRegistry | 工作如何编排? | 作战计划
**事件** | EventCatalog | 发生了什么? | 战场态势
**平台** | PlatformRegistry | 外部资源在哪? | 后勤补给
**规则** | RuleRegistry | 什么能做/不能做? | 交战规则
五个维度**各自独立又相互咬合** ——就像五角星的五个角,缺一个就不完整。
## 二、环境搭建
# 克隆 FROST 教学框架
git clone https://gitee.com/liao_liang_7514/frost.git
cd frost
# 安装依赖
pip install -r requirements.txt
# 验证环境
python -m pytest test_core.py -v
FROST 的设计哲学是**零外部依赖** ——核心只需要 Python 标准库。五维元模型的模块也遵循这个原则。
## 三、维度一:Armory(武器注册表)
Armory 管理 Agent 所有能力的元数据。不是简单的方法注册,而是带元信息的**能力目录** 。
from core.armory import Armory, SkillMetadata
# 创建武器库
armory = Armory()
# 注册一个技能(带完整元数据)
armory.register(
SkillMetadata(
name="summarize_text",
category="nlp",
description="将长文本压缩为摘要",
input_schema={"text": "string", "max_length": "int"},
output_schema={"summary": "string"},
cost_estimate=0.002, # 每次调用预估成本
latency_ms=500, # 预估延迟
tags=["summarization", "compression"]
)
)
armory.register(
SkillMetadata(
name="search_web",
category="retrieval",
description="联网搜索获取最新信息",
input_schema={"query": "string", "top_k": "int"},
output_schema={"results": "list[dict]"},
cost_estimate=0.01,
latency_ms=2000,
tags=["search", "real-time"]
)
)
# 发现可用技能
nlp_skills = armory.discover(category="nlp")
print(f"NLP 技能: {[s.name for s in nlp_skills]}")
# 输出: NLP 技能: ['summarize_text']
# 按标签发现
search_skills = armory.discover(tags=["real-time"])
print(f"实时技能: {[s.name for s in search_skills]}")
# 输出: 实时技能: ['search_web']
**关键设计** :Armory 不是简单的字典。它支持**分类发现** 、**标签过滤** 和**成本估算** ,让 Agent 在执行前就知道"我有什么武器、每把武器多少成本"。
## 四、维度二:TaskRegistry(任务注册表)
TaskRegistry 管理任务的 DAG(有向无环图)编排。任务不再是扁平列表,而是有依赖关系的图谱。
from core.task_registry import TaskRegistry, Task, TaskDependency
registry = TaskRegistry()
# 定义任务
task_collect = Task(
id="collect_data",
name="收集数据",
skill="search_web",
params={"query": "Agent治理最新研究", "top_k": 5}
)
task_analyze = Task(
id="analyze_data",
name="分析数据",
skill="summarize_text",
params={"max_length": 200}
)
task_report = Task(
id="generate_report",
name="生成报告",
skill="summarize_text",
params={"max_length": 500}
)
# 注册任务
registry.add_task(task_collect)
registry.add_task(task_analyze)
registry.add_task(task_report)
# 定义依赖关系(DAG)
registry.add_dependency(TaskDependency(
from_task="collect_data",
to_task="analyze_data"
))
registry.add_dependency(TaskDependency(
from_task="analyze_data",
to_task="generate_report"
))
# 获取执行顺序(拓扑排序)
execution_order = registry.get_execution_order()
print(f"执行顺序: {[t.id for t in execution_order]}")
# 输出: 执行顺序: ['collect_data', 'analyze_data', 'generate_report']
# 获取可并行执行的任务
parallel_tasks = registry.get_parallelizable()
print(f"可并行任务: {parallel_tasks}")
**关键设计** :TaskRegistry 把任务从"线性列表"升级为"DAG 图谱"。这意味着:
* 无依赖的任务可以**并行执行**
* 依赖关系是**显式声明** 的,不是隐式嵌套
* 支持**动态插入** 新任务到图谱中
## 五、维度三:EventCatalog(事件编目)
EventCatalog 是系统的"态势感知"中心。所有 Agent 的行为都产生事件,事件可以被订阅和分析。
from core.event_catalog import EventCatalog, AgentEvent, EventType
from core.strategist import Strategist
catalog = EventCatalog()
# 发布事件
catalog.publish(AgentEvent(
type=EventType.TASK_STARTED,
source_agent="analyst",
payload={"task_id": "collect_data", "query": "Agent治理"}
))
catalog.publish(AgentEvent(
type=EventType.TASK_COMPLETED,
source_agent="analyst",
payload={"task_id": "collect_data", "results_count": 5, "duration_ms": 1800}
))
catalog.publish(AgentEvent(
type=EventType.RULE_VIOLATION,
source_agent="executor",
payload={"rule": "max_cost_per_task", "attempted": 0.5, "limit": 0.1}
))
# 查询事件
violations = catalog.query(
event_type=EventType.RULE_VIOLATION,
time_range="1h"
)
print(f"最近1小时违规: {len(violations)} 次")
# 用 Strategist 进行态势分析
strategist = Strategist(catalog)
# 模式分析:检测异常模式
analysis = strategist.analyze_patterns(window="24h")
print(f"异常模式: {analysis.anomalies}")
# 可能输出: 异常模式: ['executor 连续3次触发成本限制']
# 统计摘要
summary = strategist.get_summary()
print(f"事件总数: {summary.total_events}")
print(f"成功率: {summary.success_rate:.1%}")
**关键设计** :EventCatalog 不是日志系统——它是**结构化的态势感知** 。每个事件有类型、来源、时间戳和载荷,支持查询、过滤和模式分析。
配合 `Strategist` 模块,系统可以自动检测异常模式(比如某个 Agent 频繁违规),实现**自治愈** 的第一步。
## 六、维度四:PlatformRegistry(平台注册表)
PlatformRegistry 管理外部平台和服务的接入。无论是 LLM API、数据库还是第三方服务,都通过统一的注册表管理。
from core.platform_registry import PlatformRegistry, PlatformConfig, HealthCheck
registry = PlatformRegistry()
# 注册 LLM 平台
registry.register(PlatformConfig(
name="deepseek",
type="llm",
endpoint="https://api.deepseek.com/v1",
auth_type="api_key",
rate_limit=100, # 每分钟100次
timeout_seconds=30,
health_check=HealthCheck(
interval_seconds=60,
endpoint="/models",
expected_status=200
)
))
# 注册向量数据库
registry.register(PlatformConfig(
name="chromadb",
type="vector_db",
endpoint="http://localhost:8000",
auth_type="none",
rate_limit=1000,
timeout_seconds=10
))
# 查询可用平台
llm_platforms = registry.list(type="llm")
print(f"可用LLM: {[p.name for p in llm_platforms]}")
# 健康检查
health = registry.check_health("deepseek")
print(f"DeepSeek状态: {health.status}")
print(f"延迟: {health.latency_ms}ms")
# 输出: DeepSeek状态: healthy, 延迟: 45ms
# 自动降级:如果主平台不可用,切换到备用
platform = registry.get_healthy_or_fallback("llm", fallback="local_ollama")
print(f"使用平台: {platform.name}")
**关键设计** :PlatformRegistry 实现了**优雅降级** ——当主平台不可用时自动切换到备用。这在生产环境中至关重要,因为没有任何外部服务是 100% 可靠的。
## 七、维度五:RuleRegistry(规则注册表)
RuleRegistry 是治理的最后一道防线。规则是**代码级别的约束** ,不是提示词级别的建议。
from core.rule_registry import RuleRegistry, Rule, RuleScope, Severity
rules = RuleRegistry()
# 注册成本规则
rules.register(Rule(
id="cost_limit_per_task",
scope=RuleScope.TASK,
severity=Severity.ERROR,
condition="cost <= 0.1",
description="单个任务成本不得超过 $0.1",
version="1.0.0"
))
# 注册数据访问规则
rules.register(Rule(
id="readonly_for_descendants",
scope=RuleScope.AGENT,
severity=Severity.CRITICAL,
condition="agent.generation > 0 => store.mode == 'readonly'",
description="后代Agent只能只读访问祖先Store",
version="1.0.0"
))
# 注册代数限制规则
rules.register(Rule(
id="max_generation_limit",
scope=RuleScope.SYSTEM,
severity=Severity.ERROR,
condition="agent.generation <= agent.max_spawn_generation",
description="Agent代数不得超过系统最大限制",
version="2.0.0"
))
# 合规检查:在执行前验证
task_cost = 0.15 # 预估成本
compliance = rules.check(
rule_id="cost_limit_per_task",
context={"cost": task_cost}
)
if not compliance.passed:
print(f"❌ 违规: {compliance.message}")
print(f" 严重程度: {compliance.severity}")
print(f" 规则版本: {compliance.rule_version}")
# 输出:
# ❌ 违规: 成本 $0.15 超过限制 $0.1
# 严重程度: ERROR
# 规则版本: 1.0.0
else:
print("✅ 合规,可以执行")
# 规则版本管理
rules.update("cost_limit_per_task", new_condition="cost <= 0.2", new_version="1.1.0")
history = rules.get_history("cost_limit_per_task")
print(f"规则版本历史: {[h.version for h in history]}")
# 输出: 规则版本历史: ['1.0.0', '1.1.0']
**关键设计** :RuleRegistry 的三个核心特性:
1. **代码强制** :违规直接抛异常,不是"建议"
2. **版本管理** :规则可以迭代,历史可追溯
3. **作用域分层** :系统级/Agent级/任务级,粒度可控
## 八、五维协同:构建完整的治理闭环
五个维度不是孤立的——它们协同工作形成治理闭环:
"""
完整的五维协同示例:一个受治理的Agent任务执行流程
"""
from core.armory import Armory
from core.task_registry import TaskRegistry
from core.event_catalog import EventCatalog
from core.platform_registry import PlatformRegistry
from core.rule_registry import RuleRegistry
# 1. 初始化五维系统
armory = Armory()
tasks = TaskRegistry()
events = EventCatalog()
platforms = PlatformRegistry()
rules = RuleRegistry()
# 2. 注册能力(Armory)
armory.register_skill("search", category="retrieval", cost=0.01)
armory.register_skill("summarize", category="nlp", cost=0.005)
armory.register_skill("write_report", category="generation", cost=0.02)
# 3. 注册平台(PlatformRegistry)
platforms.register("deepseek", type="llm", rate_limit=100)
platforms.register("chromadb", type="vector_db", rate_limit=1000)
# 4. 注册规则(RuleRegistry)
rules.register_rule("total_cost_limit", condition="total_cost <= 0.5")
rules.register_rule("platform_health", condition="platform.status == 'healthy'")
# 5. 编排任务(TaskRegistry)
tasks.add("search_task", skill="search")
tasks.add("analyze_task", skill="summarize")
tasks.add("report_task", skill="write_report")
tasks.add_dependency("search_task", "analyze_task")
tasks.add_dependency("analyze_task", "report_task")
# 6. 执行流程(五维协同)
def execute_governed_workflow():
"""受治理的工作流执行"""
# Step 1: 发布任务开始事件
events.publish("task_started", {"workflow": "research_report"})
# Step 2: 检查平台健康状态
if not platforms.is_healthy("deepseek"):
events.publish("platform_fallback", {"from": "deepseek", "to": "local"})
platforms.switch("llm", "local_ollama")
# Step 3: 获取任务执行顺序
order = tasks.get_execution_order()
total_cost = 0
for task_id in order:
# Step 4: 查询技能成本
skill = tasks.get_skill(task_id)
cost = armory.get_cost(skill)
total_cost += cost
# Step 5: 规则检查(执行前拦截)
compliance = rules.check("total_cost_limit",
context={"total_cost": total_cost})
if not compliance.passed:
events.publish("rule_violation", {
"task": task_id,
"rule": "total_cost_limit",
"total_cost": total_cost
})
raise GovernanceError(f"预算超限: ${total_cost:.3f}")
# Step 6: 执行任务
result = execute(task_id)
# Step 7: 发布完成事件
events.publish("task_completed", {
"task": task_id,
"cost": cost,
"result_length": len(str(result))
})
# Step 8: 生成治理报告
report = {
"total_tasks": len(order),
"total_cost": total_cost,
"events_generated": events.count("1h"),
"rules_violated": events.count_by_type("rule_violation", "1h")
}
events.publish("workflow_completed", report)
return report
# 执行
result = execute_governed_workflow()
print(f"✅ 工作流完成")
print(f" 任务数: {result['total_tasks']}")
print(f" 总成本: ${result['total_cost']:.3f}")
print(f" 事件数: {result['events_generated']}")
print(f" 违规数: {result['rules_violated']}")
**这个流程展示了五维协同的力量** :
* Armory 让 Agent 知道"我能做什么"
* TaskRegistry 让 Agent 知道"先做什么后做什么"
* PlatformRegistry 让 Agent 知道"资源在哪、是否可用"
* RuleRegistry 让 Agent 知道"什么不能做"
* EventCatalog 让整个系统"看到发生了什么"
## 九、FROST-SOP 中的工程实现
FROST 定义了五维元模型的理论,**FROST-SOP** 则将其落地为工程平台。
在 FROST-SOP 中,五维模型体现在以下模块:
frost-sop/
├── core/
│ ├── toolkit.py # → Armory 的工程实现(工具注册+沙箱执行)
│ ├── sop_engine.py # → TaskRegistry 的工程实现(SOP引擎)
│ ├── scheduler.py # → EventCatalog 的工程实现(事件调度)
│ ├── llm_client.py # → PlatformRegistry 的工程实现(LLM平台管理)
│ └── permission.py # → RuleRegistry 的工程实现(P0-P4权限检查)
**核心差异** :
* FROST 的 `RuleRegistry` 定义规则语法 → FROST-SOP 的 `PermissionChecker` 实现 P0-P4 五级权限
* FROST 的 `PlatformRegistry` 定义平台注册 → FROST-SOP 的 `LLMClient` 实现 Ollama 本地 + 云端双模式
* FROST 的 `TaskRegistry` 定义 DAG → FROST-SOP 的 `SOPEngine` 实现 YAML 驱动的 SOP 编排
这就是**教学框架 + 工程平台** 的分工:FROST 帮你理解"为什么这样设计",FROST-SOP 帮你实现"怎么在生产环境跑起来"。
## 十、与其他框架的对比
治理能力 | LangChain | CrewAI | FROST
---|---|---|---
能力注册表 | Tool 装饰器 | 角色绑定 | Armory(元数据+成本)
任务编排 | Chain 线性 | 角色协作 | TaskRegistry(DAG图谱)
事件系统 | 回调函数 | 无 | EventCatalog(结构化)
平台管理 | 手动配置 | 手动配置 | PlatformRegistry(自动降级)
规则约束 | 无 | 提示词 | RuleRegistry(代码强制)
FROST 不追求"什么都能做",它追求的是**"做什么都可治理"** 。
## 总结
FROST 的五维元模型不是纸上谈兵——每一个维度都有对应的代码实现和 197 个测试用例保障质量。
如果你正在构建多 Agent 系统,不妨问自己:
1. 你的 Agent 知道自己有哪些能力吗?(Armory)
2. 你的任务有显式的依赖关系吗?(TaskRegistry)
3. 你能实时看到系统发生了什么吗?(EventCatalog)
4. 外部服务挂了怎么办?(PlatformRegistry)
5. 你的约束是代码还是提示词?(RuleRegistry)
如果五个问题中有三个答不上来,也许是时候看看 FROST 了。
🔗 **项目链接**
* FROST 教学框架:https://gitee.com/liao_liang_7514/frost
* FROST-SOP 工程平台:https://gitee.com/liao_liang_7514/frost-sop
* FROST 白皮书(第四版):https://gitee.com/liao_liang_7514/frost/blob/main/WHITEPAPER.md
**FROST: 让智能体拥有谱系、记忆和荣誉感。**
_#Python #Agent #AI #OpenSource #FROST #MultiAgent #Governance_