A Python framework for building multi-role AI agent teams. Migrated from AI-Team, stripped of controller/session/web layers — keeping only the core: API clients, model providers, tools, roles, and conversation management.
AOC/
├── api/ # API backends (OpenAI, GLM, Anthropic, Gemini)
├── provider/ # Model provider definitions (27 providers)
├── tools/ # Tool implementations (37 tools)
├── roles/ # Role definitions (code + writing teams)
├── communication.py # Conversation orchestrator (Actor, Communication)
├── config/ # Config and path management
├── company_runtime/ # Event-driven multi-agent organizational governance
│ ├── schemas.py # Pydantic data models (RoleIdentity, Event, TaskNode, DailyMemoryLog)
│ ├── persistence.py # SQLite-backed durable state persistence
│ ├── event_bus.py # Three-tier async event bus with attention-gated filtering
│ ├── agent_runtime.py # Agent lifecycle: clock_in, wake_and_think, clock_out
│ └── governance.py # RBAC permission engine + budget manager
└── pyproject.toml
- Multi-provider: DeepSeek, GLM, Anthropic, OpenAI, Google, OpenRouter, xAI + 19 more
- 37 built-in tools: file I/O, shell commands, glob/grep search, browser automation, MCP support
- Multi-role: Code team (architect, coder, reviewer, security, tester, attacker) + Writing team (planner, writer, checker, polisher, condenser, researcher) + Organizer
- Inter-role delegation:
call_roletool lets roles delegate tasks to each other - MCP integration: connect to MCP servers to discover and use external tools
cd AOC
pip install -e .from roles import load_team
from communication import Actor, Communication
load_team("code")
actor = Actor("organizer")
comm = Communication.get_or_create("deepseek_v4_flash", actor)
comm.append_user_message("Write a Python hello world script")
comm.send()
print(comm.last_response)| Env Variable | Description |
|---|---|
DEEPSEEK_API_KEY |
DeepSeek API key |
GLM_API_KEY / ZHIPUAI_API_KEY |
GLM/Zhipu API key |
OPENROUTER_API_KEY |
OpenRouter API key |
ANTHROPIC_API_KEY |
Anthropic API key |
基于 asyncio + Pydantic + SQLite 的纯异步事件驱动框架，用于构建有组织结构、预算约束和权限控制的 Agent 团队。
- 绝对无轮询 — 基于
asyncio.Queue+asyncio.Event的纯推送模型 - 三层注意力过滤 — World Event Queue → Attention Gate → Cognitive Trigger
- 严格状态持久化 — clock_in / clock_out 机制，断电重启可完整恢复
- 身份注入 — 每次唤醒从结构化
RoleIdentity重建 System Prompt，不依赖长对话历史 - 预算治理 + RBAC — 每个操作前强制通过 BudgetManager 和 PermissionEngine 拦截
python -m company_runtime运行内置集成测试，展示两个 Agent（Tech Lead + Backend Engineer）的完整协作流程。
外部消息 / 定时信号
│
▼
┌────────────────────┐
│ Tier 1: EventBus │ ← asyncio.Queue 事件中心
│ World Event Queue │
└───────┬────────────┘
│ 广播到所有订阅者
▼
┌────────────────────┐
│ Tier 2: Attention │ ← 每 Agent 独立闸门
│ Gate 过滤器 │ 规则: target_id 匹配 → salience=1.0
│ │ @提及 → salience=1.0
│ salience ≥ 0.6 ? │ 职责关键词匹配 → 加权
│ YES → Inbox │ salience < 0.6 → DROP
│ NO → 丢弃 │ 去抖动: 同 event_id 2 秒内不重复
└───────┬────────────┘
│ 事件入列
▼
┌────────────────────┐
│ Tier 3: Cognitive │ ← 监听到 Inbox 有事件
│ Trigger 唤醒 │ AgentState: SLEEPING → AWAKE
│ │ 调用 wake_and_think()
└───────┬────────────┘
│
▼
AgentRuntime.wake_and_think(event)
OFF_DUTY ──clock_in()──▶ SLEEPING
SLEEPING ──inbox 有事件──▶ AWAKE ──wake_and_think() 完成──▶ SLEEPING
AWAKE ──budget exceeded──▶ BLOCKED ──approval granted──▶ AWAKE
AWAKE ──clock_out()─────▶ OFF_DUTY
PENDING ──▶ IN_PROGRESS ──▶ COMPLETED (终态)
│ │
└──▶ AWAITING_APPROVAL ◀──┘
│
├──▶ IN_PROGRESS (批准后恢复)
└──▶ FAILED (终态)
| 模块 | 文件 | 职责 |
|---|---|---|
| Schemas | schemas.py |
Pydantic 数据模型: RoleIdentity, Event, TaskNode, DailyMemoryLog, PermissionNode |
| Persistence | persistence.py |
SQLite 持久化: 每 Agent 一个 .db 文件，事件日志按天 JSONL |
| Event Bus | event_bus.py |
三层事件总线: EventBus (发布/订阅), AttentionGate (过滤), CognitiveTrigger (唤醒) |
| Agent Runtime | agent_runtime.py |
Agent 生命周期: clock_in, wake_and_think, build_system_prompt, clock_out |
| Governance | governance.py |
组织治理: BudgetManager (预算拦截+升级审批), PermissionEngine (RBAC 校验) |
import asyncio
from company_runtime import *
async def main():
# 基础设施
persistence = StatePersistence(".data")
event_bus = EventBus()
budget_mgr = BudgetManager()
perm_engine = PermissionEngine()
# 定义身份
manager = RoleIdentity(
agent_id="mgr-1",
role_title="Tech Lead",
responsibilities=["Oversee backend team", "Approve budgets"],
permissions=["approve:budget:*", "delegate:task:*"],
daily_budget_limit=50.0,
)
engineer = RoleIdentity(
agent_id="eng-1",
role_title="Backend Engineer",
responsibilities=["Build REST APIs"],
permissions=["tool:github:push"],
reports_to="mgr-1",
daily_budget_limit=2.0, # 故意设低触发升级
)
# 创建 Agent
mgr = AgentRuntime(manager, event_bus, persistence)
eng = AgentRuntime(engineer, event_bus, persistence)
# 上班
await mgr.clock_in()
await eng.clock_in()
# 派发任务
task = TaskNode(task_id="t-1", assignee_id="eng-1",
input_context={"title": "Build Payment API"})
await mgr.delegate_task("eng-1", task)
# 预算超限 → 自动升级审批 → 经理批准 → 任务恢复
await budget_mgr.check(eng, 3.0, task_id="t-1")
await asyncio.sleep(0.1) # 事件传播
await budget_mgr.approve_budget(mgr, "eng-1", "t-1", extra_budget=10.0)
# 下班落盘
off = Event(event_type=EventType.OFF_DUTY, target_id="mgr-1")
await event_bus.publish(off)
await asyncio.sleep(0.1)
# 重启后恢复
memory = persistence.load_daily_memory("mgr-1")
print(memory.summary_of_day)
asyncio.run(main())经理 (Tech Lead, $50/天) ←→ 员工 (Backend Engineer, $2/天)
1. CLOCK_IN → 双方上班，各自被自己的打卡事件唤醒
2. TASK_ASSIGNED → 经理派发 "Build Payment API v2" 任务给员工
(闸门确保经理不被自己的派发事件唤醒)
3. BUDGET_EXCEEDED → 员工 $2 预算耗尽 → BLOCKED → 升级至经理
4. APPROVAL_GRANTED → 经理批准追加 $10 → 员工恢复工作
(闸门确保经理不被自己的批准事件唤醒)
5. OFF_DUTY → 双方下班 → DailyMemoryLog 写入 SQLite
6. 验证 → 重新加载落盘数据，确认完整恢复
事件日志 (JSONL): 7 条事件，零噪音唤醒
MIT License