┌─────────────────────────────────────────────────────┐

│  👑 君主（你）                                      │

│      └── 发布任务、查看结果，不干预执行              │

│                                                     │

│  👴 祖辈（主Agent）                                 │

│      └── 常驻、全局编排、拆解任务                    │

│           ↓                                         │

│  🕵️ 斥候（侦查Agent）                              │

│      └── 外出狩猎、收集情报、快速验证                │

│           ↓                                         │

│  ⚔️ 府兵（执行Agent）                              │

│      └── 领命执行、重复劳动、汇报结果                │

│           ↑                                         │

│  📜 长老（监督Agent）                               │

│      └── 记录过程、沉淀教训、审计合规                │

└─────────────────────────────────────────────────────┘

from frost.core import Store, Agent, skill_set, skill_get



# 四个原子：

# 1. Store（记忆）—— Agent的工作空间

store = Store()



# 2. Skill（能力）—— 纯函数，无状态

def collect_daily_tasks(context):

    """收集今日任务"""

    tasks = ["写推广文章", "回复客户", "跟进项目"]

    context["tasks"] = tasks

    return context



def execute_task(context):

    """执行单个任务"""

    current_task = context.get("current_task", "")

    context["result"] = f"已完成: {current_task}"

    return context



# 3. Agent（细胞）—— 包裹记忆和能力

daily_agent = Agent("daily_worker", store, skills={

    "collect": collect_daily_tasks,

    "execute": execute_task

})



# 4. SOP（宪法）—— 定义执行顺序

result = daily_agent.run(

    sop_steps=["collect", "execute"],

    initial_context={"current_task": "写推广文章"}

)



print(result["result"])  # 输出：已完成: 写推广文章

# skills/content_generation/article_writer.py

from skills.base import BaseSkill, SkillResult

from typing import Dict, Any

import asyncio



class ArticleWriterSkill(BaseSkill):

    """文章撰写Skill - 府兵角色"""



    @property

    def name(self) -> str:

        return "article_writer"



    @property

    def description(self) -> str:

        return "根据主题撰写推广文章，支持多平台适配"



    @property

    def parameters(self) -> Dict[str, Any]:

        return {

            "type": "object",

            "properties": {

                "topic": {"type": "string", "description": "文章主题"},

                "platform": {"type": "string", "enum": ["devto", "juejin", "zhihu"]},

                "style": {"type": "string", "enum": ["technical", "casual", "story"]},

                "word_count": {"type": "integer", "default": 1500}

            },

            "required": ["topic", "platform"]

        }



    async def execute(self, topic: str, platform: str,

                     style: str = "technical",

                     word_count: int = 1500) -> SkillResult:

        """异步执行文章撰写"""



        # 调用LLM生成内容

        prompt = self._build_prompt(topic, platform, style, word_count)

        response = await self.llm_client.chat(prompt)



        return SkillResult(

            success=True,

            data={

                "title": self._extract_title(response),

                "content": response,

                "platform": platform,

                "word_count": len(response),

                "status": "draft"

            }

        )



    def _build_prompt(self, topic, platform, style, word_count):

        """根据平台调整提示词"""

        platform_hints = {

            "devto": "Write in a developer-focused, technical tone. Include code examples.",

            "juejin": "Write in Chinese. Focus on practical applications and real examples.",

            "zhihu": "Write in Chinese. Use a storytelling approach with deep insights."

        }

        return f"{platform_hints.get(platform)} Topic: {topic}. Style: {style}. ~{word_count} words."

# sops/daily_promotion.yaml

name: "daily_promotion"

description: "每日推广文章自动撰写与发布"

version: "1.0"



trigger:

  type: "schedule"

  cron: "0 11 * * *"  # 每天11:00执行



context:

  author: "神通说"

  topics_rotation:

    - Mon: "FROST深度"

    - Tue: "SOP工程"

    - Wed: "双项目联动"

    - Thu: "代码教程"

    - Fri: "行业趋势"

    - Sat: "社区故事"

    - Sun: "轻量科普"



steps:

  - name: "research_topic"

    skill: "topic_researcher"

    parameters:

      date: "${context.today}"

      rotation: "${context.topics_rotation}"

    halt_on_error: false



  - name: "write_article"

    skill: "article_writer"

    parameters:

      topic: "${research_topic.topic}"

      platform: "juejin"

      style: "technical"

    halt_on_error: true



  - name: "create_draft"

    skill: "draft_creator"

    parameters:

      platform: "juejin"

      title: "${write_article.title}"

      content: "${write_article.content}"

    halt_on_error: true



  - name: "publish_article"

    skill: "content_publisher"

    parameters:

      platform: "juejin"

      draft_id: "${create_draft.draft_id}"

    halt_on_error: false



  - name: "sync_other_platforms"

    skill: "multi_platform_syncer"

    parameters:

      platforms: ["devto", "zhihu"]

      original_content: "${write_article.content}"

      original_id: "${create_draft.draft_id}"

    halt_on_error: false



error_handling:

  on_failure: "alert_author"

  retry_count: 2

  fallback: "save_to_drafts"

# 查看执行报告

result = await engine.execute("daily_promotion", context)



print(f"状态: {result['status']}")          # completed

print(f"耗时: {result['duration_ms']}ms")   # 12450ms

print(f"Token: {result['token_usage']}")     # 3200

print(f"步骤数: {len(result['steps'])}")     # 5

任务进来 → 祖辈（主Agent）拆解 → 派给合适的子Agent执行



每日推广文章  → 斥候收集素材 → 府兵撰写 → 长老审计

客户咨询      → 斥候分类问题 → 府兵回答 → 长老记录

项目跟进      → 斥候检查进度 → 府兵更新 → 长老汇报

git clone https://gitee.com/liao_liang_7514/frost.git

cd frost

python -m pytest tests/ -v

git clone https://gitee.com/liao_liang_7514/frost-sop.git

cd frost-sop

pip install -r requirements.txt

streamlit run ui/dashboard.py