AI 原生智能体操作系统 · AI-Native Agent Operating System

Build Intelligent Agents That Think, Remember, and Evolve

^{Built on solid foundations, reimagined for the AI era.}

一句话理解

记忆 + 技能 + 心跳 + 质量 = 完整的 Agent 操作系统

核心亮点

五层记忆

业界领先的分层记忆架构

_{State → Context → Behavior → Cognitive → Core}

✅ 自动迁移
✅ 语义检索
✅ 智能压缩

多 Agent 协作

原生支持复杂多智能体场景

_{MsgHub 对话中心 · Pipeline 流水线编排}

✅ 角色分工
✅ 动态加入/退出
✅ 顺序/并行执行

闭环执行链路

从规划到交付的完整闭环

_{Planning → Execution → Auditing → Correction}

✅ 质量审计
✅ 自动纠偏
✅ 预算控制

长期运行

7×24 小时持久化服务

_{心跳调度 · Cron 任务 · 自愈能力}

✅ 定时任务
✅ 异常恢复
✅ IoT 集成

内置前端控制台 · Newzland

仓库包含一套可直接部署的 Newzland 前端(Vite 三页:landing / console / world),
位于 frontend/,由 scripts/build_frontend.sh 同步到 stickman/gateway/static/,
gateway 自动托管。

# 本地开发:并发起 gateway + Vite dev(proxy /api → 8000)
bash scripts/start_dev_fullstack.sh

# 本地预览:构建后单端口(8000)跑,生产一致
bash scripts/start_prod.sh

# 生产部署模板:见 deploy/README.md

架构 / 契约 / 三种形态见 docs/newzland-merge.md 与 deploy/README.md。

快速开始

# 安装 Stickman
pip install stickman-agent

# 或者从源码安装
git clone https://github.com/AI-said-incorrect/AgentOs.git
cd AgentOs
pip install -e ".[dev]"

# 设置 API Key (任选一个)
export DASHSCOPE_API_KEY='your-key'
export OPENAI_API_KEY='your-key'

# 运行第一个 Agent
python examples/basic/quickstart.py

📖 为什么选择 Stickman？

特性	传统框架	Stickman
架构	单次执行脚本	操作系统级长期运行
记忆	无或简单上下文窗口	五层自动迁移记忆系统
多 Agent	手动编排	MsgHub + Pipeline 原生支持
质量控制	无	完整闭环：规划→执行→审计→纠正
调度	无	Cron 心跳 + 事件总线
技能管理	硬编码工具	市场化 Skills 生态系统
可观测性	黑盒	全链路追踪 + 质量报告

[Stickman 是什么？]((#stickman-是什么)
核心能力总览
[ 五层记忆系统（深度解析）]((#-五层记忆系统深度解析)
多Agent协作系统
[ 完整闭环执行链路]((#-完整闭环执行链路)
心跳调度与长期运行
质量循环与自我纠偏
[ Skills生态系统]((#-skills生态系统)
[ 详细示例指南]((#详细示例指南)
解决的实际问题
快速开始
项目结构
为什么叫 Stickman？

Stickman 是什么？

Stickman 是一个全新的 AI原生智能体操作系统（AI-NAS）。

它不是框架，不是库，而是让AI Agent像操作系统一样运行的完整平台：

一句话总结

Stickman = Agent的大脑（记忆） + 手（Skills） + 心脏（心跳） + 良知（质量循环）

核心设计理念

理念	说明
OS思维	Agent不是一次性脚本，而是长期运行的服务
分层架构	记忆、技能、调度、质量各司其职
闭环反馈	执行→监控→评估→改进的完整循环
市场机制	Skills像App Store一样发现、推广、进化
企业级	可观测性、安全性、可扩展性内置

核心能力总览

┌─────────────────────────────────────────────────────────────┐
│                    Stickman AI-NAS                          │
├──────────┬──────────┬──────────┬──────────┬────────────────┤
│  🧠 记忆 │  🔧 技能 │  ⏰ 调度 │  🎯 质量 │   🤝 多Agent    │
│          │          │          │          │                │
│ •五层架构│ •市场化  │ •Cron任务│ •偏差检测│ •MsgHub对话     │
│ •自动迁移│ •信任策略│ •事件总线│ •自动纠正│ •Pipeline流水线  │
│ •语义检索│ •经验演进│ •异常告警│ •质量报告│ •角色分工       │
│ •智能压缩│ •自动合成│ •自愈能力│ •状态机  │ •动态加入退出   │
└──────────┴──────────┴──────────┴──────────┴────────────────┘

五层记忆系统（深度解析）

Stickman实现了业界领先的五层记忆架构，模拟人类认知过程：

架构设计

用户输入 → State(状态) → Context(上下文) → Behavior(行为) → Cognitive(认知) → Core(核心)
   ↑            ↓              ↓               ↓              ↓           ↓
   └────────────┴──────────────┴───────────────┴──────────────┴───────────┘
                              自动迁移引擎

各层详解

层级	类比人类	存储内容	存储方式	TTL	用途
State	工作记忆	当前会话状态、临时变量	Redis	24h	会话管理、实时追踪
Context	短期记忆	对话上下文、近期交互	PostgreSQL+向量	7天	语义理解、上下文连贯
Behavior	中期记忆	行为模式、偏好习惯	PostgreSQL+Redis	30天	个性化、行为预测
Cognitive	长期知识	领域知识、推理规则	Neo4j+Milvus	永久	知识图谱、复杂推理
Core	人格核心	身份认同、价值观、长期目标	PostgreSQL	永久	人格一致性、长期规划

自动迁移机制

from stickman.memory.tiered_memory import TieredMemoryManager

manager = TieredMemoryManager()

# 添加记忆 - 系统会根据重要性自动分配层级
await manager.add_memory(
    content="用户喜欢用Python做数据分析",
    importance_score=0.8,        # 重要性评分 (0-1)
    tags=["preference", "tech"], # 标签
    metadata={
        "source": "conversation",
        "timestamp": "2026-04-17"
    }
)

# 迁移触发条件：
# 1. 访问频率高 → 向上层迁移（更重要）
# 2. 时间衰减 → 向下层迁移或删除
# 3. 重要性评分高 → 保留在高层级
# 4. 容量满时 → 触发压缩

迁移算法核心要素：

LRU (最近最少使用)：频繁访问的记忆优先保留
时间衰减指数：decay = e^(-λ * Δt)，λ为半衰期参数
重要性加权：score = frequency × importance × decay
容量阈值：每层独立配置max_size

语义检索

# 自然语言查询，跨层检索
results = await manager.search(
    query="编程语言相关偏好",
    top_k=5,
    filters={"tags": ["preference"]}  # 可选过滤
)

for result in results:
    print(f"层级: {result.layer}, 相关度: {result.score:.3f}")
    print(f"内容: {result.content}")

智能压缩

当某层容量达到阈值时，系统自动进行无损压缩：

原始记忆（10条）:
1. [10:00] 用户问：Python怎么处理CSV？
2. [10:01] 我回答了pandas用法
3. [10:05] 用户又问了Excel读取
4. [10:06] 推荐了openpyxl
5. [10:15] 用户想合并两个DataFrame
6. ...

压缩后摘要（1条）:
[10:00-10:30] 数据处理对话：
  - 用户关注Python数据处理（CSV、Excel、DataFrame）
  - 推荐：pandas（CSV）、openpyxl（Excel）
  - 用户技能水平：中级，需要具体代码示例

压缩策略：

时间窗口分组：按时间段组织相关记忆
角色感知：区分用户输入、系统响应、工具调用结果
关键信息提取：保留实体、决策、结论、数值
损失函数优化：最大化信息保留率，最小化存储空间

使用示例

import asyncio
from stickman.agent import ReActAgent
from stickman.model import DashScopeChatModel
from stickman.formatter import DashScopeChatFormatter
from stickman.memory import TieredMemoryManager
from stickman.memory._tiered_memory.in_memory_layers import (
    InMemoryStateLayer,
    InMemoryContextLayer,
    InMemoryBehaviorLayer,
    InMemoryCognitiveLayer,
    InMemoryCoreLayer,
)


async def main():
    # 创建五层记忆
    tiered_memory = TieredMemoryManager(
        state_layer=InMemoryStateLayer(max_size=2),
        context_layer=InMemoryContextLayer(max_size=3),
        behavior_layer=InMemoryBehaviorLayer(max_size=4),
        cognitive_layer=InMemoryCognitiveLayer(max_size=5),
        core_layer=InMemoryCoreLayer(),
    )

    # 创建带分层记忆的Agent
    agent = ReActAgent(
        name="Friday",
        sys_prompt="你是一个有记忆的智能助手。",
        model=DashScopeChatModel(
            api_key=os.getenv("DASHSCOPE_API_KEY"),
            model_name="qwen-max",
        ),
        formatter=DashScopeChatFormatter(),
        memory=tiered_memory,  # 注入分层记忆
    )

    # 对话中记忆自动工作
    response = await agent("我叫小明，我喜欢Python")
    # → 存入State层
    
    response = await agent("我刚才说我喜欢什么？")
    # → 从Context层检索到"Python"
    
    response = await agent("记住我的生日是2020-01-01")
    # → 根据重要性存入Core层（永久保存）

asyncio.run(main())

完整示例位置: examples/memory/tiered_memory_agent_example.py

多Agent协作系统

Stickman原生支持多Agent场景，从简单对话到复杂工作流：

MsgHub - 对话中心

MsgHub 是多Agent通信的核心枢纽：

from stickman.pipeline import MsgHub, sequential_pipeline
from stickman.agent import ReActAgent
from stickman.message import Msg


async def multi_agent_conversation():
    # 创建不同角色的Agent
    alice = ReActAgent(
        name="Alice",
        sys_prompt="你是一位30岁的教师，性格友好。",
        model=model,
        formatter=formatter,
    )
    
    bob = ReActAgent(
        name="Bob", 
        sys_prompt="你是一位14岁的学生，性格叛逆。",
        model=model,
        formatter=formatter,
    )
    
    charlie = ReActAgent(
        name="Charlie",
        sys_prompt="你是一位28岁的医生，性格深思熟虑。",
        model=model,
        formatter=formatter,
    )

    # 在MsgHub中进行多方对话
    async with MsgHub(
        participants=[alice, bob, charlie],
        announcement=Msg("system", "请互相自我介绍。", "system"),
    ) as hub:
        # 顺序执行：每个Agent依次发言
        await sequential_pipeline([alice, bob, charlie])
        
        # 动态管理：Bob离开
        hub.delete(bob)
        await hub.broadcast(Msg("bob", "我有事先走了！", "assistant"))
        
        # 剩余Agent继续对话
        await alice()
        await charlie()

特性：

广播消息：所有参与者同时收到
点对点：指定接收者
动态加入/退出：运行时增减Agent
历史记录：完整对话日志
角色隔离：每个Agent独立记忆和人格

Pipeline - 流水线编排

将多个Agent串联成工作流：

from stickman.pipeline import sequential_pipeline, parallel_pipeline


# 顺序流水线：A → B → C
await sequential_pipeline([
    researcher_agent,      # 第一步：研究
    writer_agent,         # 第二步：撰写  
    reviewer_agent,       # 第三步：审核
])

# 并行流水线：A,B,C 同时执行
results = await parallel_pipeline([
    translator_en,       # 英文翻译
    translator_zh,       # 中文翻译
    summarizer,          # 摘要生成
])

典型应用场景

场景1：辩论赛 (`examples/game/werewolves/`)

# 狼人杀游戏 - 12个Agent角色互动
players = [
    WerewolfAgent(name="狼人1", role="werewolf"),
    VillagerAgent(name="村民1", role="villager"),
    SeerAgent(name="预言家", role="seer"),
    WitchAgent(name="女巫", role="witch"),
    # ... 更多角色
]

async with MsgHub(participants=players) as game_hub:
    for round_num in range(10):
        # 白天讨论
        await sequential_pipeline(players)
        # 投票放逐
        vote_result = await run_voting(game_hub)
        # 夜晚行动
        await night_phase(game_hub)

场景2：深度研究 (`examples/integration/qwen_deep_research_model/`)

# 多阶段研究流程
research_pipeline = [
    PlannerAgent("制定研究计划"),
    SearchAgent("搜索资料"), 
    AnalysisAgent("分析数据"),
    SynthesisAgent("综合报告"),
    ReviewAgent("质量审核"),
]

场景3：多语言服务

# 同一请求并行处理多种语言
async with MsgHub() as service_hub:
    en_agent = TranslatorAgent(target_lang="en")
    zh_agent = TranslatorAgent(target_lang="zh")
    ja_agent = TranslatorAgent(target_lang="ja")
    
    results = await parallel_pipeline([en_agent, zh_agent, ja_agent])

完整闭环执行链路

Stickman提供从任务接收到结果交付的完整闭环：

执行流程图

┌─────────────────────────────────────────────────────────────────┐
│                    任务输入 (User Request)                       │
└────────────────────────────┬────────────────────────────────────┘
                             ▼
┌─────────────────────────────────────────────────────────────────┐
│  Phase 1: 理解与规划 (Planning)                                  │
│  • 解析用户意图                                                  │
│  • 分解子任务                                                    │
│  • 选择合适的Skills                                              │
│  • 制定执行计划                                                  │
└────────────────────────────┬────────────────────────────────────┘
                             ▼
┌─────────────────────────────────────────────────────────────────┐
│  Phase 2: 执行与工具调用 (Execution)                             │
│  • 调用大模型推理                                                 │
│  • 使用Tools/Skills                                             │
│  • 访问记忆系统                                                   │
│  • 收集中间结果                                                  │
└────────────────────────────┬────────────────────────────────────┘
                             ▼
┌─────────────────────────────────────────────────────────────────┐
│  Phase 3: 质量审计 (Auditing)                                    │
│  • 检查输出完整性                                                │
│  • 评估结果准确性                                                │
│  • 检测偏差和异常                                                │
│  • 生成质量分数                                                  │
└────────────────────────────┬────────────────────────────────────┘
                             ▼
              ┌──────────────┴──────────────┐
              ▼                             ▼
┌─────────────────────┐         ┌─────────────────────────────┐
│ 通过 ✓              │         │ 未通过 ✗                     │
│                     │         │                             │
▼                     │         ▼                             │
┌─────────────────────┐  ┌─────────────────────────────┐
│ Phase 4: 输出交付   │  │ 偏差分析与纠正               │
│ • 格式化结果        │  │ • 识别问题根因              │
│ • 更新记忆          │  │ • 选择纠正策略              │
│ • 记录执行日志      │  │ • 重新执行                  │
│ • 通知用户          │  │ • 或请求人工干预            │
└─────────────────────┘  └──────────────┬──────────────┘
                                        ▼
                              回到Phase 2重试
                              （最多N轮）

质量循环实现

from stickman.quality_loop import (
    QualityLoopOrchestratorV2,
    QualityExecutorAgent,
    QualityAuditorAgent,
    QualityControllerAgent,
    RoundBudget,
)


async def quality_controlled_execution():
    # 配置预算约束
    budget = RoundBudget(
        max_rounds=3,           # 最多尝试3轮
        max_failures=2,         # 最多失败2次
        max_cost_per_round=1.0, # 每轮最大成本($)
        max_latency_ms=5000,    # 最大延迟(ms)
    )

    # 创建执行代理
    executor = QualityExecutorAgent(
        name="executor",
        sys_prompt="完成业务任务，返回结构化结果。",
        model=model,
        memory=memory,
        toolkit=toolkit,
    )

    # 创建审计代理
    auditor = QualityAuditorAgent(
        name="auditor", 
        sys_prompt="严格评估执行结果质量。",
        model=model,
        memory=memory,
    )

    # 创建控制代理
    controller = QualityControllerAgent(
        name="controller",
        sys_prompt="决定是否继续、完成或终止。",
        model=model,
        deviation_config={...},
        budget=budget,
    )

    # 编排器协调整个流程
    orchestrator = QualityLoopOrchestratorV2(
        executor=executor,
        auditor=auditor,
        controller=controller,
        budget=budget,
    )

    # 执行任务（自动经过完整闭环）
    result = await orchestrator.execute(user_request)
    
    return {
        "output": result.final_output,
        "quality_score": result.quality_score,
        "rounds_completed": result.rounds_used,
        "cost_total": result.total_cost,
    }

示例位置: examples/workflows/quality_loop/demo_v2.py

闭环的关键优势

传统方案	Stickman闭环
一次生成，无检查	多轮迭代，持续优化
无质量标准	量化评分(0-100)
失败就报错	自动重试+降级
黑盒执行	全链路可观测
无成本控制	预算约束+提前终止

心跳调度与长期运行

解决的问题

传统Agent脚本：

执行完就退出
无法定时任务
无法持久化状态
异常无法恢复

Stickman心跳调度：

7×24小时运行
Cron表达式调度
状态持久化
自愈能力

使用示例

from stickman.heartbeat.scheduler import HeartbeatScheduler
from stickman.heartbeat.event_bus import EventBus
from stickman.heartbeat.alert import AlertSystem


async def long_running_agent():
    scheduler = HeartbeatScheduler()
    event_bus = EventBus()
    alert_system = AlertSystem()

    # 注册定时任务
    scheduler.add_cron_job(
        func=daily_health_check,
        cron_expression="0 9 * * *",  # 每天9点
        job_id="daily_check"
    )

    scheduler.add_cron_job(
        func=weekly_report,
        cron_expression="0 10 * * 1",  # 每周一10点
        job_id="weekly_summary"
    )

    # 监听事件
    @event_bus.on("anomaly_detected")
    async def handle_anomaly(data):
        alert_system.send_alert(
            severity="warning",
            message=f"检测到异常: {data}",
            channel=["email", "slack"]
        )
        
        # 自动修复
        if data["auto_fixable"]:
            await auto_repair(data)

    # 启动调度器
    await scheduler.start()

    # 保持运行...
    await asyncio.Event().wait()

示例位置: examples/heartbeat/basic_usage.py, advanced_features.py

IoT硬件集成

from stickman.iot.edge import EdgeDevice


# 连接物理设备
device = EdgeDevice(port="COM3", baudrate=9600)

async def monitor_sensor():
    while True:
        # 读取传感器数据
        temperature = await device.read_sensor("temp")
        humidity = await device.read_sensor("humidity")
        
        # 存入记忆
        await memory.add({
            "type": "sensor_data",
            "temperature": temperature,
            "humidity": humidity,
            "timestamp": datetime.now()
        })
        
        # 异常检测
        if temperature > 40:
            await event_bus.emit("overheating", {"temp": temperature})
        
        await asyncio.sleep(60)  # 每分钟采集

质量循环与自我纠偏

核心组件

Quality Loop Architecture:

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  Executor    │────>│   Auditor    │────>│  Controller  │
│  (执行代理)  │     │  (审计代理)  │     │ (控制代理)   │
└──────────────┘     └──────────────┘     └──────────────┘
       │                   │                   │
       ▼                   ▼                   ▼
  执行业务任务        评估输出质量        决定下一步动作
  调用Tools          检测偏差             ✓ 继续/✗ 重试/终止
  收集证据           生成分数             成本控制

偏差检测类型

类型	检测方法	处理策略
事实错误	与知识库比对	自动纠正或标记
格式不符	Schema验证	重新格式化
逻辑矛盾	上下文一致性	要求解释
安全风险	敏感词扫描	拦截并告警
性能超标	延迟/成本监控	降级或终止

配置示例

config = {
    "deviation_threshold": 0.15,      # 偏差容忍度
    "auto_correct": True,             # 自动纠正
    "human_review_threshold": 0.3,    # 超过此值需人工审核
    "max_retry": 3,                   # 最大重试次数
    "fallback_mode": "safe_response", # 降级模式
}

Skills生态系统

Skill生命周期

发现(Discovery) → 注册(Register) → 测试(Test) 
→ 推广(Promote) → 稳定(Stable) → 演进(Evolve)
     ↑                                            │
     └──────────── 自动合成 ←─────────────────────┘

创建自定义Skill

目录结构：

my_skill/
├── SKILL.md          # 技能文档（必须）
├── skill.py          # 实现（可选，可纯文档）
├── meta_reminder.py  # 元提醒（可选）
└── config.json       # 配置（可选）

SKILL.md模板：

---
name: System Performance Monitor
description: 10分钟系统性能监控与报告生成
version: 1.0.0
category: monitoring
---

# 功能描述
收集CPU、内存、磁盘等指标，生成性能报告。

# 使用步骤
1. 初始化: python monitor_performance.py --init
2. 检测: python monitor_performance.py --check --round N
3. 提醒: python meta_reminder.py --status completed_round_N
4. 报告: python monitor_performance.py --report

# 参数
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| rounds | int | 否 | 检测轮数(默认11) |

# 输出
- performance_data.json: 原始数据
- performance_report.md: 分析报告

注册和使用

from stickman.tool import Toolkit

toolkit = Toolkit()

# 注册Skill目录
toolkit.register_agent_skill("path/to/my_skill")

# Agent自动发现并使用
agent = ReActAgent(
    name="monitor",
    sys_prompt="使用可用Skills完成任务。",
    toolkit=toolkit,
    # ...
)

# Agent会自动读取SKILL.md并按步骤执行
response = await agent("监控系统性能10分钟")

Skill市场特性

发现: 自动扫描注册中心
信任等级: Security Audit → Trusted → Verified
使用统计: 调用次数、成功率、平均耗时
经验演进: 基于反馈自动优化prompt
合成: 组合现有Skills生成新Skill

示例位置: test/system_performance_monitor_skill/ (完整可运行示例)

详细示例指南

基础入门

1. 简单对话Agent

文件: examples/agent/react_agent/main.py

cd examples/agent/react_agent
export DASHSCOPE_API_KEY='your-key'
python main.py

学习要点: ReAct模式、基本配置、流式输出

2. 记忆系统使用

文件: examples/functionality/short_term_memory/reme/

cd examples/functionality/short_term_memory/reme
python short_term_memory_example.py

学习要点: 短期记忆、Reme实现、记忆检索

3. 自定义Skill开发

文件: examples/functionality/agent_skill/main.py

cd examples/functionality/agent_skill
python main.py

学习要点: Skill创建、注册、Agent集成

进阶实战

4. 多Agent对话

文件: examples/workflows/multiagent_conversation/main.py

cd examples/workflows/multiagent_conversation
export DASHSCOPE_API_KEY='your-key'
python main.py

功能: 3个不同角色Agent在MsgHub中对话
学习要点: MsgHub、sequential_pipeline、动态管理

5. 质量循环演示

文件: examples/workflows/quality_loop/demo_v2.py

cd examples/workflows/quality_loop
export DASHSCOPE_API_KEY='your-key'
python demo_v2.py

功能: Executor-Auditor-Controller三代理协作
学习要点: 质量闭环、预算控制、自动重试

6. 深度研究Agent

文件: examples/integration/qwen_deep_research_model/

cd examples/integration/qwen_deep_research_model
export DASHSCOPE_API_KEY='your-key'
python main.py

功能: 多步骤研究流程（规划→搜索→分析→综合）
学习要点: 子任务分解、长程推理、报告生成

7. RAG知识库

文件: examples/functionality/rag/

cd examples/functionality/rag
# 基础用法
python basic_usage.py
# 与ReAct集成
python react_agent_integration.py
# 多模态（图片+PDF）
python multimodal_rag.py

学习要点: 文档加载、向量存储、混合检索

8. 语音交互

文件: examples/agent/voice_agent/ 或 realtime_voice_agent/

cd examples/agent/voice_agent
export DASHSCOPE_API_KEY='your-key'
python main.py

学习要点: TTS语音合成、实时语音API

高级应用

9. 狼人杀游戏

文件: examples/game/werewolves/

cd examples/game/werewolves
export DASHSCOPE_API_KEY='your-key'
python main.py

功能: 12个Agent角色扮演，完整游戏逻辑
学习要点: 复杂状态机、角色AI、多轮博弈

10. A2A协议Agent

文件: examples/agent/a2a_agent/

cd examples/agent/a2a_agent
python setup_a2a_server.py  # 启动服务器
python main.py              # 启动Agent

学习要点: A2A协议、Agent卡片、远程调用

11. 浏览器自动化

文件: examples/agent/browser_agent/

cd examples/agent/browser_agent
python main.py

功能: Agent控制浏览器完成任务
学习要点: 网页理解、表单填写、任务分解

12. 心跳调度完整示例

文件: examples/heartbeat/full_e2e_with_agent_skill.py

cd examples/heartbeat
python full_e2e_with_agent_skill.py

功能: 心跳+Agent+Skill+记忆完整集成
学习要点: 长期运行、定时任务、状态持久化

解决的实际问题

问题分类与解决方案

问题类别	具体痛点	Stickman解决方案
记忆缺失	Agent忘记之前的对话	五层记忆+自动迁移
单次执行	无法长期运行任务	心跳调度+状态持久化
质量不稳定	输出结果不可靠	质量循环+自动纠偏
技能有限	只能做预定义任务	Skills市场+自动发现
单打独斗	复杂任务无法拆解	多Agent协作+Pipeline
黑盒执行	不知道内部过程	全链路追踪+可观测性
安全风险	可能执行危险操作	沙箱隔离+权限控制
成本失控	API调用费用过高	预算约束+提前终止

适用场景

企业场景

客服系统: 有记忆的多轮对话，自动工单流转
数据分析: 定时报告生成，异常自动告警
研发辅助: Code Review、Bug分析、文档生成
运维监控: 7×24系统巡检，自动故障排查

个人场景

个人助理: 日程管理、信息整理、习惯跟踪
学习伙伴: 知识问答、学习计划、进度跟踪
创作辅助: 写作、翻译、润色、风格模仿
生活管家: 购物比价、旅行规划、健康提醒

创新场景

游戏NPC: 有记忆、有性格的游戏角色
虚拟偶像: 长期运营、粉丝互动、内容生成
科研助手: 文献综述、实验设计、数据分析
教育导师: 个性化教学、自适应难度、学情分析

快速开始

安装

# 方式1: PyPI安装
pip install stickman-agentos

# 方式2: 源码安装
git clone https://github.com/AI-said-incorrect/AgentOs.git
cd AgentOs
pip install -e .

# 方式3: 开发模式（推荐贡献者）
pip install -e ".[dev]"

最小化示例

import asyncio
from stickman.agent import ReActAgent
from stickman.model import OpenAIChatModel
from stickman.memory import InMemoryMemory
from stickman.tool import Toolkit, execute_python_code
from stickman.formatter import OpenAIChatFormatter
from stickman.message import Msg


async def main():
    # 1. 创建工具集
    toolkit = Toolkit()
    toolkit.register_tool_function(execute_python_code)
    
    # 2. 创建Agent
    agent = ReActAgent(
        name="Friday",
        sys_prompt="你是一个有帮助的AI助手。",
        model=OpenAIChatModel(
            model_name="gpt-4",
            api_key="your-api-key",
        ),
        formatter=OpenAIChatFormatter(),
        memory=InMemoryMemory(),
        toolkit=toolkit,
    )
    
    # 3. 对话
    response = await agent(Msg(role="user", content="你好！", name="user"))
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

使用通义千问（国内推荐）

from stickman.model import DashScopeChatModel
from stickman.formatter import DashScopeChatFormatter
import os

agent = ReActAgent(
    name="Assistant",
    model=DashScopeChatModel(
        model_name="qwen-max",
        api_key=os.environ["DASHSCOPE_API_KEY"],
        stream=True,
    ),
    formatter=DashScopeChatFormatter(),
    memory=InMemoryMemory(),
    toolkit=toolkit,
)

使用StepFun（已测试通过）

import os

from stickman.model import OpenAIChatModel
from stickman.formatter import OpenAIChatFormatter

agent = ReActAgent(
    model=OpenAIChatModel(
        model_name=os.environ.get("STEPFUN_MODEL_NAME", "step-3.6"),
        api_key=os.environ["STEPFUN_API_KEY"],
        client_kwargs={
            "base_url": "https://api.stepfun.com/v1"
        },
    ),
    formatter=OpenAIChatFormatter(),
    # ...
)

AgentOs gateway 的 /api/runs 也可以通过环境变量直接使用 StepFun:

export AINAS_MODEL_PROVIDER=stepfun
export STEPFUN_MODEL_NAME=step-3.6
export STEPFUN_BASE_URL=https://api.stepfun.com/v1
export STEPFUN_API_KEY=your-stepfun-api-key
python scripts/gateway_server.py --host 0.0.0.0 --port 8000 --allowed-root /data

项目结构

stickman/
├── agent/                    # Agent引擎
│   ├── _react_agent.py      # ReAct模式（思考-行动-观察循环）
│   ├── _user_agent.py       # 用户Agent（模拟用户输入）
│   ├── _realtime_agent.py   # 实时Agent（语音/视频流）
│   └── _a2a_agent.py        # A2A协议Agent
│
├── memory/                   # 记忆系统（核心亮点）
│   ├── _tiered_memory/      # ⭐ 五层分层记忆
│   │   ├── base_classes.py  # 层基类
│   │   ├── tiered_memory_manager.py  # 迁移管理器
│   │   ├── memory_compression.py     # 智能压缩
│   │   ├── in_memory_layers.py       # 内存实现
│   │   ├── postgres_layers.py        # PostgreSQL实现
│   │   └── redis_layers.py           # Redis实现
│   ├── _working_memory/     # 工作记忆（短期）
│   │   ├── _in_memory_memory.py
│   │   ├── _redis_memory.py
│   │   └── _sqlalchemy_memory.py
│   └── _long_term_memory/   # 长期记忆
│       ├── _mem0/           # Mem0实现
│       └── _reme/           # Reme实现
│           ├── _reme_long_term_memory_base.py
│           ├── _reme_personal_memory.py   # 个人记忆
│           ├── _reme_task_memory.py       # 任务记忆
│           └── _reme_tool_memory.py       # 工具记忆
│
├── model/                    # 模型适配器
│   ├── _dashscope_model.py   # 通义千问
│   ├── _openai_model.py      # OpenAI GPT
│   ├── _anthropic_model.py   # Anthropic Claude
│   ├── _gemini_model.py      # Google Gemini
│   ├── _deepseek_model.py    # DeepSeek
│   ├── _ollama_model.py      # Ollama本地
│   └── _trinity_model.py     # 三合一模型
│
├── tool/                     # 工具与Skill框架
│   ├── _toolkit.py           # 工具包管理
│   ├── _agent_skill/         # Agent Skill支持
│   ├── _coding/              # 代码执行工具
│   │   ├── _python.py
│   │   └── _shell.py
│   ├── _sandbox/             # 安全沙箱
│   └── _text_file/           # 文件操作
│
├── skills/                   # Skills生态
│   ├── market.py             # 技能市场
│   ├── trust_policy.py       # 信任策略
│   ├── experience.py         # 经验演进
│   ├── synthesis.py          # 自动合成
│   └── registry.py           # 技能注册
│
├── heartbeat/                # 心跳调度器（长期运行）
│   ├── scheduler.py          # Cron调度
│   ├── event_bus.py          # 事件总线
│   ├── alert.py              # 告警系统
│   ├── storage.py            # 状态持久化
│   └── hardware.py           # IoT硬件接口
│
├── quality_loop/             # 质量循环（闭环执行）
│   ├── orchestrator.py       # 主编排器
│   ├── orchestrator_v2.py    # V2版本（推荐）
│   ├── agents.py             # Executor/Auditor/Controller
│   ├── deviation.py          # 偏差检测
│   ├── state_machine.py      # 状态机
│   └── reporter.py           # 质量报告
│
├── pipeline/                 # 流水线与多Agent协作
│   ├── _chat_room.py         # 聊天室
│   ├── _class.py             # Pipeline基类
│   ├── _functional.py        # 函数式Pipeline
│   └── _msghub.py            # ⭐ MsgHub（消息中心）
│
├── rag/                      # RAG知识库
│   ├── _reader/              # 文档读取（PDF/Word/PPT/Excel/图片）
│   ├── _store/               # 向量存储（Milvus/Qdrant/MongoDB/MySQL）
│   └── _knowledge_base.py    # 知识库管理
│
├── mcp/                      # MCP协议支持
├── a2a/                      # A2A协议支持
├── realtime/                 # 实时通信（语音/视频）
├── tts/                      # 语音合成
├── embedding/                # 向量嵌入
├── sandbox/                  # 安全沙箱（Docker隔离）
├── iot/                      # IoT硬件集成
├── evaluate/                 # 评估框架（ACE Benchmark）
├── tuner/                    # 模型调优
├── session/                  # 会话管理
├── tracing/                  # 全链路追踪
└── gateway/                  # API网关

简单，但不简陋。

^{就像火柴人可以画出任何东西一样，Stickman 让你构建任何类型的智能 Agent}

支持的服务

大语言模型

模型提供商	支持的模型	状态
DashScope	Qwen-Max, Qwen-Plus, Qwen-Turbo
OpenAI	GPT-4o, GPT-4 Turbo, o1
Anthropic	Claude 3.5 Sonnet, Claude 3 Opus
Google	Gemini Pro, Gemini Flash
DeepSeek	V3, Coder, R1
Ollama	Llama, Mistral, Qwen (本地)
StepFun	Step-3.5-Flash

能力扩展

能力类型	支持的服务
向量嵌入	DashScope, OpenAI, Gemini, Ollama
语音合成 TTS	CosyVoice, Gemini TTS, OpenAI TTS
实时通信	DashScope Realtime, Gemini Live API, OpenAI Realtime
知识库 RAG	Milvus, Qdrant, MongoDB, OceanBase, MySQL
协议支持	MCP (Model Context Protocol), A2A (Agent-to-Agent)
硬件集成	IoT串口设备、传感器、执行器

测试与验证

⚡ 快速验证命令

# 1. 快速验证（无需API Key - 推荐先跑这个）
cd test && python test_skill_framework.py

# 2. 完整端到端测试（需要API Key）
export DASHSCOPE_API_KEY='your-key'  # Linux/Mac
$env:DASHSCOPE_API_KEY='your-key'   # Windows PowerShell
cd test && python test_system_performance_skill.py

# 3. 长程Skills测试（模拟50轮对话）
cd test && python test_real_long_term_skills.py

# 4. 记忆系统单元测试
cd test/memory && python test_tiered_memory.py

# 5. 心跳调度测试
cd test/heartbeat && python test_task_events.py

# 6. 安全沙箱测试
cd test/security && python test_sandbox_tool_routing.py

✅ 已验证通过的完整链路

测试日期: 2026-04-17
测试环境: Windows 11 · Python 3.12 · StepFun step-3.5-flash

能力	状态	说明
大模型接入		StepFun/DashScope/OpenAI
Agent自主决策		ReAct模式正常工作
工具调用		Shell/Python/文件操作
Skill注册执行		系统性能监控Skill
记忆系统集成		InMemoryMemory正常
性能数据采集		CPU/内存/磁盘/进程数
元技能提醒		进度跟踪正常
报告生成		JSON+Markdown双格式
多轮对话		2轮完整执行（Round 1 & 2）

测试详情: 参见 test/test_system_performance_skill.py 运行日志

文档资源

文档	说明	目标读者
PRD_AI_NAS.md	产品需求文档 - 规格、路线图	产品经理、架构师
ARCHITECTURE_DESIGN.md	技术架构 - 模块设计、部署	开发者、运维
TEST_CASE_GUIDE.md	测试指南 - 用例编写规范	测试工程师

贡献指南

我们欢迎所有形式的贡献！无论是代码、文档还是建议。

🚀 快速贡献流程

# 1. Fork 并克隆
git clone https://github.com/<your-username>/AgentOs.git
cd AgentOs

# 2. 创建功能分支
git checkout -b feature/amazing-feature

# 3. 开发并测试
# 编写代码 → 运行测试 → 确保通过

# 4. 提交更改
git commit -m 'feat: amazing feature'

# 5. 推送并创建 PR
git push origin feature/amazing-feature
# 在 GitHub 上提交 Pull Request

开发规范

规范项	要求
Python 版本	3.10+
代码风格	PEP 8 (使用 black 格式化)
注释语言	中文（函数用途、参数说明、返回值）
Commit 规范	Conventional Commits (`feat/fix/docs/refactor/test/chore`)
测试要求	新功能必须有对应测试，位于 `test/` 目录

🛠️ 开发环境设置

# 克隆仓库
git clone https://github.com/AI-said-incorrect/AgentOs.git
cd AgentOs

# 创建虚拟环境
python -m venv venv
venv\Scripts\activate  # Windows
source venv/bin/activate  # Linux/Mac

# 安装依赖
pip install -e ".[dev]"

# 设置API Key（任选一个）
export DASHSCOPE_API_KEY='your-key'
export OPENAI_API_KEY='your-key'

# 运行测试
python test/test_skill_framework.py

许可证

本项目基于 Apache License 2.0 开源协议。详见 LICENSE 文件。

如果这个项目对你有帮助，请给一个 Star！

^{Built with ❤️ by the Stickman Team · Powered by AI-NAS Architecture}

贡献者

贡献者	贡献内容
llwxy	核心记忆系统架构：时序知识图谱(Graphiti)、Ebbinghaus遗忘曲线(MemoryBank)、记忆关联网络(A-MEM)、跨层去重(Mem0)、Token预算管理(MemGPT)、注入反馈(LangMem)

用户18 · 2026 年5 月 10 日 15:59

牛大佬，

NeoSpecies · 2026 年5 月 11 日 12:07

没有没有大家一起进步

AI 原生智能体操作系统 · AI-Native Agent Operating System

Table of Contents

Stickman