用代码理解 FROST：AI Agent 的家族式治理架构
如果你正在设计一个 AI Agent 系统，你是否曾经被这些问题困扰：
- 如何让 Agent 具备记忆传承能力？
- 如何实现任务的自动分发与调度？
- 如何建立清晰的角色分工与监督机制？
FROST（Family-based Reasoning Operating SysTem）用生物学隐喻给出了优雅的答案。本文将通过代码示例，带你深入理解 FROST 的家族式治理架构。
一、FROST 的核心理念
FROST 不是传统意义上的工具库，而是一个「会成长的 AI Agent 家族治理框架」。它的设计哲学来自一个朴素的问题：
「如果 AI Agent 像一个家族一样运作，会怎样？」
在这个隐喻中：
祖辈是家族中唯一常驻的成员，负责全局调度
府兵是按需召唤的执行单元，完成任务后消散
长老是监督者，自动监控全流程
记忆在家族中代代传承，新成员自动继承所有经验
二、核心代码结构
FROST 的核心代码约 500 行，由三个原子类构成：
\`python
frost/core.py - 核心抽象
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
from datetime import datetime
import json
import os
class Store:
"""
记忆容器 - 家族的记忆库
负责持久化存储和检索家族的所有经验
"""
def init(self, storage_path: str = "./frost_memory"):
self.storage_path = storage_path
self.memory: Dict[str, Any] = {}
os.makedirs(storage_path, exist_ok=True)
self._load_memory()
def _load_memory(self):
"""启动时加载历史记忆"""
memory_file = os.path.join(self.storage_path, "family_memory.json")
if os.path.exists(memory_file):
with open(memory_file, \r, encoding=\utf-8) as f:
self.memory = json.load(f)
def save(self, key: str, value: Any):
"""保存记忆"""
self.memory[key] = {
"value": value,
"timestamp": datetime.now().isoformat()
}
self._persist()
def recall(self, key: str) -> Optional[Any]:
"""召回记忆"""
return self.memory.get(key, {}).get("value")
def _persist(self):
"""持久化到磁盘"""
memory_file = os.path.join(self.storage_path, "family_memory.json")
with open(memory_file, \w, encoding=\utf-8) as f:
json.dump(self.memory, f, ensure_ascii=False, indent=2)
class Skill(ABC):
"""
技能 - 纯函数变换
每个技能都是独立的、可以组合的变换单元
"""
@abstractmethod
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""执行技能，返回变换后的上下文"""
pass
@property
@abstractmethod
def name(self) -> str:
"""技能名称"""
pass
class Agent:
"""
执行单元 - 家族成员
具备特定角色的执行能力，可以是斥候、军师或府兵
"""
def init(self, role: str, skills: List[Skill], store: Store):
self.role = role
self.skills = skills
self.store = store
self.context: Dict[str, Any] = {}
def receive_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
"""接收任务"""
self.context = {
"task": task,
"role": self.role,
"start_time": datetime.now().isoformat()
}
return self.context
def execute(self) -> Dict[str, Any]:
"""按顺序执行所有技能"""
for skill in self.skills:
self.context = skill.execute(self.context)
self.context["end_time"] = datetime.now().isoformat()
return self.context
`\
三、家族治理流程实现
\`python
frost/governance.py - 家族治理
from typing import List, Callable
from dataclasses import dataclass
from enum import Enum
class Role(Enum):
"""家族角色枚举"""
SENTRY = "斥候" # 情报收集
STRATEGIST = "军师" # 策略制定
EXECUTOR = "府兵" # 执行交付
ELDER = "长老" # 监督审计
@dataclass
class Task:
"""任务描述"""
name: str
description: str
priority: int = 1
@dataclass
class TaskResult:
"""任务结果"""
task: Task
output: Dict[str, Any]
status: str
audit_notes: List[str]
class FamilyGovernance:
"""
家族治理中枢
实现祖辈的核心职责：任务分发与监督
"""
def init(self, store: Store):
self.store = store
self.agents: Dict[Role, List[Agent]] = {
Role.SENTRY: [],
Role.STRATEGIST: [],
Role.EXECUTOR: [],
Role.ELDER: []
}
self.execution_log: List[TaskResult] = []
def register_agent(self, role: Role, agent: Agent):
"""注册家族成员"""
self.agents[role].append(agent)
print(f"✅ 家族新成员注册: {agent.__class__.__name__} 担任 {role.value}")
def dispatch_task(self, task: Task) -> TaskResult:
"""
任务分发核心逻辑
1. 派斥候收集情报
2. 派军师制定策略
3. 派府兵执行
4. 长老审计
"""
print(f"📋 祖辈开始分发任务: {task.name}")
# Step 1: 斥候狩猎情报
sentry_output = self._execute_role(Role.SENTRY, task)
# Step 2: 军师制定策略
strategist_output = self._execute_role(Role.STRATEGIST, task, sentry_output)
# Step 3: 府兵执行交付
executor_output = self._execute_role(Role.EXECUTOR, task, strategist_output)
# Step 4: 长老审计
audit_notes = self._elder_audit(task, executor_output)
result = TaskResult(
task=task,
output=executor_output,
status="completed" if not audit_notes else "completed_with_issues",
audit_notes=audit_notes
)
self.execution_log.append(result)
self._save_experience(task, result)
return result
def _execute_role(self, role: Role, task: Task, context: Dict = None) -> Dict:
"""执行特定角色的任务"""
agents = self.agents.get(role, [])
if not agents:
return {"status": "no_agent", "message": f"无 {role.value} 可用"}
agent = agents[0]
agent.receive_task({
"task": task,
"context": context or {}
})
return agent.execute()
def _elder_audit(self, task: Task, output: Dict) -> List[str]:
"""长老自动审计"""
notes = []
if "start_time" in output and "end_time" in output:
notes.append(f"执行时长已记录")
if not output.get("result"):
notes.append("⚠️ 输出结果为空，需要关注")
return notes
def _save_experience(self, task: Task, result: TaskResult):
"""经验自动沉淀到记忆库"""
self.store.save(
f"task_{task.name}_{datetime.now().date()}",
{
"task_name": task.name,
"status": result.status,
"output_summary": str(result.output)[:200],
"lessons": result.audit_notes
}
)
print(f"📚 经验已沉淀到家族记忆库")
`\
四、快速上手示例
\`python
example/task_execution.py
from frost.core import Store, Skill, Agent
from frost.governance import FamilyGovernance, Task, Role
class InformationGatherer(Skill):
@property
def name(self) -> str:
return "information_gatherer"
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
task = context.get("task", {})
context["intelligence"] = {
"scope": f"为任务 \{task.name} 收集的相关信息",
"data_points": ["数据源A", "数据源B", "数据源C"]
}
return context
class StrategyPlanner(Skill):
@property
def name(self) -> str:
return "strategy_planner"
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
context["strategy"] = {
"approach": "分阶段执行",
"phases": ["准备", "执行", "验证"],
"risk_points": ["依赖项", "时间窗口"]
}
return context
class TaskExecutor(Skill):
@property
def name(self) -> str:
return "task_executor"
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
context["result"] = {
"status": "success",
"deliverables": ["产出物1", "产出物2"],
"metrics": {"完成度": "100%", "质量": "优秀"}
}
return context
def setup_family():
store = Store("./my_family_memory")
governance = FamilyGovernance(store)
sentry = Agent(role="斥候", skills=[InformationGatherer()], store=store)
governance.register_agent(Role.SENTRY, sentry)
strategist = Agent(role="军师", skills=[StrategyPlanner()], store=store)
governance.register_agent(Role.STRATEGIST, strategist)
executor = Agent(role="府兵", skills=[TaskExecutor()], store=store)
governance.register_agent(Role.EXECUTOR, executor)
return governance
if name == "main":
family = setup_family()
task = Task(name="每日推广文章", description="撰写并发布 FROST 推广内容", priority=1)
result = family.dispatch_task(task)
print(f"\n📊 任务完成状态: {result.status}")
print(f"📝 审计备注: {result.audit_notes}")
`\
运行输出：
\`
✅ 家族新成员注册: Agent 担任 斥候
✅ 家族新成员注册: Agent 担任 军师
✅ 家族新成员注册: Agent 担任 府兵
📋 祖辈开始分发任务: 每日推广文章
📚 经验已沉淀到家族记忆库
📊 任务完成状态: completed
📝 审计备注: [\执行时长已记录]
`\
五、FROST 的独特优势
1. 记忆代际传承
每次府兵执行完成后，经验自动沉淀到 Store。新成员启动时，自动加载全量家族记忆，无需重复踩坑。
2. 角色清晰分工
斥候 → 军师 → 府兵 → 长老，形成天然的任务流水线，每个角色专注做好一件事。
3. 极简设计哲学
核心仅 500 行代码，通过组合而非继承实现扩展，新技能即插即用。
4. 监督闭环
长老自动审计每个任务，确保质量可追溯、问题可复盘。
六、下一步
想深入了解 FROST？
Gitee: https://gitee.com/liao_liang_7514/frost
文档: 完整的 API 文档和更多示例
关于作者
本文是 FROST「公开造物」系列的第 16 篇，完整记录了用 FROST 方法论构建自身的过程。如果你也感兴趣，欢迎 fork 项目，一起探索 AI Agent 的家族式治理之道。
FROST - 让 AI Agent 像家族一样成长
SOCIAL SHARE CARD GENERATOR