用 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 的三个核心特性：
- 代码强制：违规直接抛异常，不是"建议"
- 版本管理：规则可以迭代，历史可追溯
- 作用域分层：系统级/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 系统，不妨问自己：
- 你的 Agent 知道自己有哪些能力吗？（Armory）
- 你的任务有显式的依赖关系吗？（TaskRegistry）
- 你能实时看到系统发生了什么吗？（EventCatalog）
- 外部服务挂了怎么办？（PlatformRegistry）
- 你的约束是代码还是提示词？（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
Top comments (0)