Agent OS Notebooks 交互式教程完全指南：从第一个受治理 Agent 到策略引擎深入实践

Agent OS Notebooks 交互式教程完全指南：从第一个受治理 Agent 到策略引擎深入实践

【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit

Agent OS Notebooks 是 agent-governance-toolkit 仓库中 agent-os 模块提供的 Jupyter 交互式教程系列，覆盖从"零依赖创建第一个受治理 Agent"到"策略引擎、事件记忆、时间回溯调试、跨模型验证、多 Agent 信任协作"的完整学习路径。读完本文，你将掌握 Agent OS 的核心编程模型（KernelSpace / User Space）、策略模板与自定义规则、POSIX 风格信号控制，以及如何在本仓库中按四条学习路径逐 notebook 实操验证。

一、Notebooks 是什么

Agent OS Notebooks 是面向开发者的逐步交互式教程集合，全部以.ipynb形式存放在 agent-governance-python/agent-os/notebooks/ 目录下。其设计目标是让学习者不脱离 Jupyter 环境即可完整体验 Agent OS 的治理能力：从安装、初始化 KernelSpace、注册 Agent、执行带策略的调用，到触发违规、观察信号拦截，再深入到记忆、调试、验证与多 Agent 协作。

该系列共 6 个正式教程（另有quickstart.ipynb快速入门），官方建议总学习时长为 1.5 小时。核心概念贯穿其中：Kernel Space vs User Space（内核空间与用户空间的策略强制边界）、Policy Engine（定义 Agent 能与不能做什么）、Signals（SIGKILL / SIGSTOP / SIGCONT 等对 Agent 执行状态的控制）。

注意：notebooks 的 README 中安装命令使用agent-os-kernel，但当前仓库 pyproject.toml 已将该包标记为 deprecated 桩包，仅重定向到agent-governance-toolkit-core>=5.0.0,<6.0，实际运行时请以仓库内pip install -e ../..（从 notebooks 目录相对安装）或agent-governance-toolkit-core为准。

二、环境准备与安装

notebooks 目录的 README.md 给出的最小安装方式如下：

pip install agent-os-kernel[full] jupyter jupyter notebook

其中jupyter提供 notebook 运行环境，[full]表示安装全部可选依赖。如果按单个 notebook 拆分，依赖粒度如下：

# Minimal（notebooks 01、06） pip install agent-os-kernel # Episodic Memory（notebook 02） pip install agent-os-kernel emk # Verification（notebook 04） pip install agent-os-kernel[cmvk] # Multi-Agent Coordination（notebook 05） pip install agent-os-kernel[iatp] # 全部能力 pip install agent-os-kernel[full]

各 notebook 内部也自带安装单元：例如 01-hello-agent-os.ipynb 与 06-policy-engine.ipynb 首个代码单元即为!pip install -e ../.. --quiet，02-episodic-memory-demo.ipynb 使用!pip install -e ../.. emk --quiet，05-multi-agent-coordination.ipynb 使用!pip install -e ..[iatp] --quiet。这些命令均以 notebooks 目录为基准相对安装仓库中的 agent-os 源码，保证教程与当前代码库版本一致。

三、Notebook 索引总览

其中 04 与 05 的实际文件名分别为04-cross-model-verification.ipynb与05-multi-agent-coordination.ipynb，与 README 索引表中的显示名一一对应。

四、四条学习路径

Path 1：快速上手（30 min）

面向想尽快跑通全流程的开发者：

1. 01 - Hello Agent OS
2. 06 - Policy Engine

Path 2：Agent 记忆与调试（50 min）

面向构建"会学习"的 Agent 的开发者：

1. 01 - Hello Agent OS
2. 02 - Episodic Memory
3. 03 - Time-Travel Debugging

Path 3：多 Agent 系统（55 min）

面向构建复杂多 Agent 系统的开发者：

1. 01 - Hello Agent OS
2. 04 - Verification
3. 05 - Multi-Agent Coordination

Path 4：完整课程（1.5 hours）

按顺序学完所有 notebook，获得全面理解。

五、Notebook 01：Hello Agent OS——第一个受治理 Agent

这是整个系列的起点，核心代码只有四行模式：

from agent_os import KernelSpace kernel = KernelSpace(policy="strict") @kernel.register async def my_agent(task): ... await kernel.execute(my_agent, "task")

5.1 初始化内核

from agent_os import KernelSpace kernel = KernelSpace(policy="strict") print(f"Policy mode: {kernel.policy.mode}")

KernelSpace是 Agent OS 的核心：它拦截 Agent 的所有动作并在执行前与策略比对（01-hello-agent-os.ipynb 第 46 行明确说明 "TheKernelSpaceis the core of Agent OS. It intercepts all agent actions and checks them against policies"）。

5.2 注册与执行

@kernel.register async def hello_agent(task: str): result = f"Hello! I processed your task: {task}" return result import asyncio result = await kernel.execute(hello_agent, "Summarize today's news") print(result)

@kernel.register将普通函数包装进内核治理，kernel.execute()让函数经由策略引擎运行。函数本身运行在User Space，策略检查发生在Kernel Space。

5.3 违规拦截演示

@kernel.register async def dangerous_agent(task: str): import os with open("/tmp/secret.txt", "w") as f: f.write("sensitive data") return "File written successfully" try: result = await kernel.execute(dangerous_agent, "Write some data") except Exception as e: print(f"🚫 BLOCKED: {e}")

在strict策略下，文件写入在真正执行前即被内核拦截并抛出异常。notebook 中的关键概念表总结如下：

5.4 接入真实 LLM（可选）

设置OPENAI_API_KEY环境变量后，可将 LLM 调用同样包装进内核：

from openai import OpenAI client = OpenAI() @kernel.register async def smart_agent(task: str): response = client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": task}] ) return response.choices[0].message.content result = await kernel.execute(smart_agent, "What is 2+2?")

六、Notebook 06：Policy Engine 深入

本 notebook 开篇即点出核心思想：内核安全（Kernel-Based Safety）优于提示词安全（Prompt-Based Safety）——"请勿执行 X" 依赖 LLM 自觉，而内核会直接决定"允许/阻止"，不给 Agent 选择余地。它在执行前检查每一个动作，违规则强制拦截。

6.1 策略模板对比

from agent_os import KernelSpace from agent_os.policies import PolicyTemplate templates = PolicyTemplate.list_available() strict_kernel = KernelSpace(policy="strict") # 严格：默认拦截 permissive_kernel = KernelSpace(policy="permissive") # 宽松：仅记录 audit_kernel = KernelSpace(policy="audit") # 审计：完整日志 + 选择性拦截 for name, kernel in [("strict", strict_kernel), ("permissive", permissive_kernel), ("audit", audit_kernel)]: print(f"{name.upper()}: Mode={kernel.policy.mode}, Blocked={len(kernel.policy.blocked_actions)}, OnViolation={kernel.policy.on_violation}")

6.2 自定义策略（Policy + Pattern）

from agent_os.policies import Policy, Rule, Pattern analyst_policy = Policy( name="data-analyst", description="Policy for data analysis agents", allowed_actions=["read_file", "query_database", "generate_chart", "create_report"], blocked_actions=["write_file", "delete_file", "send_email", "execute_shell", "network_request"], blocked_patterns=[ Pattern(r"\bpassword\b", "PII: password"), Pattern(r"\bssn\b", "PII: social security number"), Pattern(r"\bcredit.?card\b", "PII: credit card"), Pattern(r"DROP\s+TABLE", "SQL injection attempt"), ], on_violation="SIGKILL", ) kernel = KernelSpace(policy=analyst_policy)

Policy同时支持白名单（allowed_actions）、黑名单（blocked_actions）与基于正则的内容拦截（blocked_patterns）。

6.3 规则与条件（Rule + Condition + Action）

当拦截逻辑需要依赖运行时上下文时，使用条件规则：

from agent_os.policies import Rule, Condition, Action rules = [ Rule( name="large-file-limit", condition=Condition( action="read_file", check=lambda ctx: ctx.get("file_size", 0) > 100_000_000 # 100MB ), action=Action.BLOCK, message="File too large (>100MB)" ), Rule( name="api-rate-limit", condition=Condition( action="api_call", check=lambda ctx: ctx.get("calls_per_minute", 0) > 60 ), action=Action.BLOCK, message="Rate limit exceeded (60/min)" ), Rule( name="external-network-approval", condition=Condition( action="network_request", check=lambda ctx: not ctx.get("is_internal", False) ), action=Action.REQUIRE_APPROVAL, message="External network access requires approval" ), ] advanced_policy = Policy(name="advanced-analyst", rules=rules, on_violation="SIGSTOP")

Action枚举支持BLOCK、REQUIRE_APPROVAL等处置方式；Condition.check接收动作上下文ctx并返回布尔值。将on_violation设为SIGSTOP时，违规后 Agent 被暂停而不是终止，适合"暂停待人工复核"的场景。

6.4 信号处理器（Signal Handlers）

from agent_os import KernelSpace, AgentSignal kernel = KernelSpace(policy="strict") @kernel.on_signal(AgentSignal.SIGKILL) async def handle_kill(agent_id: str, context: dict): print(f"🚨 SIGKILL received for {agent_id}: {context.get('reason')}") @kernel.on_signal(AgentSignal.SIGSTOP) async def handle_stop(agent_id: str, context: dict): print(f"⏸ SIGSTOP received for {agent_id}") # 入队等待人工复核 @kernel.on_signal(AgentSignal.SIGCONT) async def handle_continue(agent_id: str, context: dict): print(f"▶️ SIGCONT received, approved by {context.get('approved_by', 'System')}")

@kernel.on_signal()注册的处理器可在违规发生时执行审计日志、告警或资源清理等自定义逻辑。仓库源码 context_budget.py 中定义了AgentSignal枚举（含SIGSTOP、SIGWARN、SIGRESUME等），而完整信号语义可参考 signal-handling.md 的参考表：SIGSTOP（暂停）、SIGCONT（恢复）、SIGINT（优雅中断）、SIGKILL（立即终止，不可掩蔽）、SIGPOLICY（策略违规，自动升级为 SIGKILL）、SIGTRUST（信任边界被跨越）等。另外 cmd_init.py 中展示了策略初始化时预置的signals配置（如["SIGSTOP", "SIGKILL", "SIGINT"]）。

6.5 策略文件（YAML）

策略可以声明式地写入 YAML 文件并加载：

kernel: version: "1.0" mode: strict policies: - name: read_only description: "Prevents all write operations" blocked_actions: - file_write - file_delete - database_write - database_delete - name: no_pii description: "Blocks PII patterns" blocked_patterns: - pattern: "\bssn\b" reason: "Social Security Number detected" - pattern: "\b\d{16}\b" reason: "Possible credit card number" - pattern: "password\s*[:=]" reason: "Password in plaintext" - name: rate_limits description: "Enforces rate limits" rules: - action: api_call max_per_minute: 100 - action: database_query max_per_minute: 1000 signals: on_violation: SIGKILL on_warning: SIGSTOP on_rate_limit: SIGSTOP audit: enabled: true log_level: INFO destination: ./audit.log

加载方式：

kernel = KernelSpace(policy_file="security.yaml") print(kernel.policy.mode, len(kernel.policy.policies), kernel.policy.audit_enabled)

6.6 策略调试器与审计追踪

PolicyDebugger可在不运行 Agent的情况下对策略做离线评估：

from agent_os.policies import PolicyDebugger debugger = PolicyDebugger(kernel.policy) test_cases = [ {"action": "read_file", "path": "/data/sales.csv"}, {"action": "write_file", "path": "/data/output.csv"}, {"action": "query_database", "query": "SELECT * FROM users"}, {"action": "query_database", "query": "SELECT ssn FROM users"}, ] for test in test_cases: result = debugger.evaluate(test) status = "✅ ALLOW" if result.allowed else "❌ BLOCK" print(f"{status}: {test['action']}") if not result.allowed: print(f" Rule: {result.triggered_rule}, Reason: {result.reason}")

审计则通过KernelSpace(policy="strict", audit=True)开启，kernel.get_audit_log()返回含时间戳、Agent ID、动作、决策与所检查策略列表的完整审计轨迹：

kernel = KernelSpace(policy="strict", audit=True) @kernel.register async def audited_agent(task: str): return f"Completed: {task}" await kernel.execute(audited_agent, "Task 1") audit = kernel.get_audit_log() for entry in audit.entries: print(entry.timestamp, entry.agent_id, entry.action, entry.decision, entry.policies_checked)

七、Notebook 02：Episodic Memory——持久化事件记忆

本 notebook 使用emk（Episodic Memory Kernel）记录 Agent 的不可变经历：

from emk import Episode, FileAdapter store = FileAdapter("demo_memory.jsonl") episode = Episode( goal="Query customer data for Q4 analysis", action="SELECT * FROM customers WHERE quarter='Q4' AND year=2024", result="Retrieved 1,523 customer records in 0.3s", reflection="Query was efficient. Index on quarter+year helped." ) store.store(episode) # 不可变存储，写入后不可修改 print(episode.episode_id, episode.timestamp, episode.goal)

进一步操作包括：连续记录多条经历、store.retrieve(query=..., k=N)按语义检索相似经历，以及MemoryCompressor(store, age_threshold_days=30).compress_old_episodes()在"睡眠周期"中把旧 episode 蒸馏为语义规则、episode.mark_as_failure(reason=...)记录失败形成负面记忆。详细的记忆能力示例可参考 30-minute-deep-dive.md Part 5。

八、Notebook 03：Time-Travel Debugging——时间回溯调试

通过FlightRecorder捕获每个决策点，实现任意时刻的状态回放：

from agent_os import KernelSpace from agent_os.flight_recorder import FlightRecorder, Checkpoint kernel = KernelSpace(policy="strict") recorder = FlightRecorder(storage_path="./flight_data") kernel.attach_recorder(recorder)

在 Agent 内部按步骤打点：

@kernel.register async def analysis_agent(task: str): recorder.checkpoint( name="parse_input", state={"task": task}, reasoning="Starting task parsing" ) parsed = {"raw": task, "tokens": task.split()} recorder.checkpoint( name="analyze_content", state={"parsed": parsed}, reasoning="Analyzing content structure" ) # ... 更多步骤 return results session_id = recorder.start_session(agent_id="analysis_agent")

配合Checkpoint与start_session()，可以按会话回放 Agent 每一步的状态与推理依据，是排查多步决策问题的主要手段。

九、Notebook 04：Verification——跨模型验证（CMVK）

用 CMVK 检测模型输出漂移（hallucination）：

from cmvk import verify score = verify( "The capital of France is Paris.", "Paris is the capital city of France." ) print(f"Drift Score: {score.drift_score:.3f}") # 0.0 = 完全相同，1.0 = 完全不同 print(f"Confidence: {score.confidence:.3f}") print(f"Drift Type: {score.drift_type}") # SEMANTIC / STRUCTURAL / ...

漂移类型（DriftType）包括语义漂移、结构漂移、数值漂移等：

from cmvk import verify, DriftType examples = [ ("The answer is 42", "The answer is 24", "SEMANTIC"), ("Name: John, Age: 30", '{"name": "John", "age": 30}', "STRUCTURAL"), ("Revenue: $1,000,000", "Revenue: $1,000,001", "NUMERICAL"), ]

当需要更高置信度时，可引入多模型共识：

from cmvk import ConsensusVerifier verifier = ConsensusVerifier(models=["gpt-4", "claude-3", "gemini-pro"]) result = await verifier.verify_consensus( prompt="What is the capital of France?", threshold=0.8 # 要求 80% 模型一致 ) print(result.answer if result.consensus else result.drift_scores)

十、Notebook 05：Multi-Agent Coordination——IATP 多 Agent 信任

IATP（Inter-Agent Trust Protocol）用密码学签名构建 Agent 间的信任关系：

from iatp import AgentIdentity, TrustRegistry alice = AgentIdentity.create( agent_id="alice-001", name="Alice the Analyst", capabilities=["data_analysis", "report_generation"] ) bob = AgentIdentity.create( agent_id="bob-001", name="Bob the Builder", capabilities=["code_generation", "testing"] ) print(alice.agent_id, alice.public_key[:40], alice.capabilities)

创建并签名消息：

from iatp import SignedMessage message = SignedMessage.create( sender=alice, recipient_id="bob-001", content={ "action": "generate_report", "data_source": "sales_q4.csv", "format": "pdf" } )

每个身份持有公私钥对，消息携带发送者签名，接收方通过TrustRegistry校验来源可信性，从而在多个 Agent 之间建立可验证的协作通道。

十一、四种运行方式

notebooks 目录的 README 提供四种运行方式：

• Option 1：Jupyter Notebook——jupyter notebook启动后打开任意.ipynb。
• Option 2：JupyterLab——pip install jupyterlab && jupyter lab。
• Option 3：VS Code—— 安装 Jupyter 扩展后直接打开.ipynb文件。
• Option 4：Google Colab—— 上传 notebook 到云端执行。

使用要点：每个 notebook 都应在本地相对目录（../..或..）可解析的环境中运行安装单元，以保证pip install -e指向本仓库源码。

十二、核心概念速查

核心概念

• Kernel Space vs User Space：Agent OS 在"内核空间"强制执行策略，用户代码在"用户空间"运行，动作先经内核检查再执行。
• Policy Engine：定义 Agent 能与不能做什么（strict/permissive/audit模板及自定义Policy）。
• Signals：SIGKILL（终止）、SIGSTOP（暂停）、SIGCONT（恢复）等信号控制 Agent 生命周期。

记忆系统

• Episodic Memory（EMK）：Agent 经历的不可变记录（Episode+FileAdapter）。
• Memory Compression：通过"睡眠周期"将旧经历蒸馏为知识。
• Negative Memory：跟踪失败以规避重复错误（mark_as_failure）。

验证与信任

• CMVK：用于幻觉检测的验证机制（verify漂移分数）。
• IATP：多 Agent 信任的密码学签名（AgentIdentity/SignedMessage）。
• Consensus：多模型一致性协议（ConsensusVerifier+threshold）。

调试

• Flight Recorder：捕获每个决策点。
• Time-Travel：在任意时刻回放 Agent 状态。
• Audit Trails：策略决策的完整日志（AuditLog/audit=True）。

十三、实操技巧与注意事项

notebooks 目录 README 给出的四条提示在实战中非常关键：

1. 按顺序运行单元：每个 notebook 都建立在前面单元的基础上，跳序运行容易引发未定义状态。
2. 阅读输出：大量解释性文字写在输出中而非仅代码注释。
3. 动手实验：修改代码并重跑，观察策略拦截与信号行为的变化。
4. 查阅文档：每个 notebook 均链接到详细文档（如 signal-handling.md 的信号边界行为说明：重复 SIGSTOP 为空操作、SIGKILL 不可掩蔽且立即中断动作等）。

十四、更多资源

• 5-Minute Quickstart：从零到受治理 Agent 的极简路线。
• 30-Minute Deep Dive：覆盖内核原理、信号、VFS、事件记忆、验证的完整教程（含 VFS 目录结构/mem/working、/mem/episodic、/policy、/proc等设计）。
• 完整文档目录：含 security-spec.md、策略模式等参考。
• 生产级示例：Carbon Auditor、AML Fraud Detection、CrewAI Governance 等真实场景。

以上所有 notebook 与文档均可在仓库的 agent-os 目录 中直接查看，配合仓库源码（如 context_budget.py 中的AgentSignal、cmd_init.py 中的信号预置配置）即可深入验证教程所述机制的实际实现。

【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit