用代码理解 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 像家族一样成长