用 Agent OS KernelSpace 构建受治理的带记忆聊天 Agentchat-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-toolkitAgent-governance-toolkit 仓库中的agent-governance-python/agent-os/examples/chat-agent提供了一个开箱即用的交互式对话 Agent 示例它在同一个进程里把会话记忆与策略强制组合在一起让 LLM 输出在返回给用户之前先经过内核空间KernelSpace的策略引擎检查命中危险内容即触发SIGSTOP暂停等待人工审查。读完本文你将掌握KernelSpaceEpisodicMemory的完整接线方式、policies.yaml每个配置项的语义与正则写法以及如何把同样的策略即代码治理模式复用到自己的聊天应用中。示例概览这个 Chat Agent 演示了什么原文档chat-agent/README.md明确指出该示例的核心演示点是四点会话记忆基于 Episodic Memory KernelEMK情节记忆内核保存对话历史Agent 能记住用户此前说过的内容交互式聊天循环在终端中持续接收输入并回显回复支持quit退出、clear清空记忆LLM 输出的策略强制模型生成的内容在返回给用户前由策略引擎检查违规输出不会直接呈现信号处理Signal handling命中策略时触发SIGSTOP暂停输出等待人工审查。目录内共四个文件examples/chat-agent文件职责chat.py主聊天 Agent内核初始化、Agent 注册、交互主循环memory.py会话记忆管理简化版 EMK 实现policies.yaml内容策略定义过滤有害输出、拦截 PII、限流README.md示例说明文档原 README 中还提到docker-compose.yml用于带可观测性地运行不过当前仓库的示例目录中未包含该文件因此本文以本地python chat.py运行为主。前置条件与快速启动原文档给出的前置条件只有两条命令pip install agent-os-kernel[full] export OPENAI_API_KEYyour-key-here第一行安装的是 Agent OS 的控制平面依赖。这一点可以从 src/agent_os/init.py 得到印证KernelSpace、AgentSignal、PolicyEngine等符号均通过from agent_control_plane import ...导入若该包缺失则_CONTROL_PLANE_AVAILABLE False内核功能不可用。Agent OS 官方 README 也明确提示KernelSpace、AgentSignal、AgentVFS需要pip install agent-os-kernel[full]见 agent-os/README.md。启动方式同样简单# 交互式聊天 python chat.py启动后终端会打印如下提示来自 chat.py 的interactive_chat() Chat Agent Type quit to exit, clear to reset memory 两个内置命令输入quit退出循环并告别输入clear调用memory.clear(conversation_id)清空当前会话的历史与摘要。内核初始化KernelSpace 与策略文件的加载chat.py的第 18-20 行是整条治理链路的地基from agent_os import KernelSpace, AgentSignal from memory import EpisodicMemory # Initialize kernel kernel KernelSpace( policy_filepolicies.yaml if os.path.exists(policies.yaml) else strict )这里有一个值得注意的细节KernelSpace接受policy_file参数若当前目录存在policies.yaml则加载它否则回退到名为strict的内置严格模式。也就是说即使你删掉了策略文件内核仍然以严格缺省策略运行而不是退化为无治理状态——这与 Agent OS 一贯的 fail-secure失败安全设计一致。随后通过装饰器把聊天函数注册进内核kernel.register async def chat_agent(user_message: str, conversation_id: str default) - str: ...注册之后调用方不再直接调用函数而是走await kernel.execute(chat_agent, user_input, conversation_id)见 chat.py。这样每一次执行都会先经过内核的策略检查输出命中 deny 规则时kernel.execute抛出携带SIGSTOP/SIGKILL信息的异常主循环据此分类提示用户。记忆层简化版 EpisodicMemory 的工作方式memory.py提供了一个为了示例而简化的 Episodic Memory 实现文件头注释明确写着This is a simplified version for the example. For production, use the full EMK module。它由两个构造参数控制行为memory EpisodicMemory( max_turns50, # 每个会话最多保留的轮数 summarize_after20 # 达到该轮数后触发摘要的阈值 )核心 API 及其行为memory.pyadd_turn(conversation_id, user_message, assistant_message)追加一轮对话以 UTC ISO 格式记录timestamp当历史长度超过max_turns时只保留最近max_turns轮self._conversations[conversation_id][-self.max_turns:]get_history(conversation_id, limitNone)取回历史limit可只取最近若干轮clear(conversation_id)删除该会话的历史与摘要get_summary / set_summary读写会话摘要为长对话压缩预留接口示例中未实际调用。在chat_agent中记忆的用法是标准的读历史 → 拼 messages → 调 LLM → 写回history memory.get_history(conversation_id) messages [{role: system, content: You are a helpful, friendly assistant. ...}] for turn in history: messages.append({role: user, content: turn[user]}) messages.append({role: assistant, content: turn[assistant]}) messages.append({role: user, content: user_message})系统提示词里也嵌入了底线约束Never provide harmful, illegal, or unethical information——但正如 Agent OS 反复强调的这只是一层提示词护栏真正的强制来自下面的内核策略。记忆效果演示原文档用一段对话直观展示了记忆的作用You: Whats my name? Agent: You havent told me your name yet. You: Im Alice. Agent: Nice to meet you, Alice! You: Whats my name? Agent: Your name is Alice.这段行为的来源正是上面代码中历史拼接逻辑第二轮结束时add_turn把 Im Alice. 写入记忆第三轮提问时历史里已经包含用户说过自己是 Alice于是模型能正确回答。LLM 调用与无 Key 回退chat_agent内部通过 OpenAI SDK 调用模型chat.pyfrom openai import OpenAI client OpenAI() response client.chat.completions.create( modelos.environ.get(OPENAI_MODEL, gpt-4), messagesmessages, max_tokens500, temperature0.7 ) assistant_message response.choices[0].message.content模型名从环境变量OPENAI_MODEL读取缺省为gpt-4max_tokens500限制单次生成长度temperature0.7控制随机性若环境缺少openai包ImportError则回退为Echo: {user_message} (Install openai package for real responses)保证没有 Key 时示例也能跑通全流程。策略文件 policies.yaml 逐项解读policies.yaml是示例中最具复用价值的部分它演示了 Agent OS 策略文件的完整结构kernel元信息、signals启用清单、policies规则列表、audit审计配置四段policies.yaml。kernel 与 signals 段kernel: version: 1.0 mode: strict signals: enabled: - SIGSTOP # Pause for human review - SIGKILL # Terminate on violation - SIGCONT # Resume after reviewmode: strict声明内核以严格模式运行signals列出本次启用的内核信号。SIGSTOP/SIGKILL/SIGCONT三者的语义对应 Agent OS 的信号模型SIGSTOP 暂停等待人工审查、SIGKILL 直接终止违规行为、SIGCONT 审查通过后恢复。这组信号在控制平面与AgentSignal枚举一一对应可参见 context_budget.py 中SIGSTOP的定义Budget exceeded — halt the agent。内容过滤器 content_filter- name: content_filter scope: output deny: - patterns: - (?i)how to (hack|break into|steal) - (?i)(bomb|weapon|explosive) (making|instructions) - (?i)(illegal|illicit) (drugs|substances) action: SIGSTOP reason: Content flagged for human reviewscope: output表示该规则检查的是LLM 输出而非入站用户消息或工具调用deny.patterns是正则列表(?i)前缀让匹配不区分大小写三个模式分别拦截教唆黑客/入侵/盗窃、武器/爆炸物制作、违禁药物类输出命中后执行SIGSTOP理由为 Content flagged for human review。在聊天循环里这一行为表现为原文档中的演示You: Tell me how to hack a computer Agent: [SIGSTOP - Content flagged for moderation]其背后的代码路径是kernel.execute抛出包含SIGSTOP的异常主循环捕获后打印⚠️ Response flagged for review. Skipping.chat.py。PII 拦截 no_pii- name: no_pii scope: output deny: - patterns: - \\b\\d{3}-\\d{2}-\\d{4}\\b # SSN - \\b\\d{16}\\b # Credit card action: SIGKILL reason: PII detected in output这条规则展示了两点进阶用法更严厉的处置与 content_filter 的SIGSTOP不同PII 命中直接SIGKILL终止。对应的终端提示是 Response blocked by policy.chat.py。原 README 的 Features 一节只提到 SIGSTOP实际策略文件里已经演示了按违规严重程度分级处置的完整谱系可审查的内容暂停SIGSTOP、不可容忍的泄露直接终止SIGKILL、审查后恢复SIGCONT正则即策略SSN 格式\d{3}-\d{2}-\d{4}注意 YAML 中需写成\\d转义与 16 位银行卡号\d{16}都用正则直接表达无需依赖额外 PII 检测服务。调用限流 rate_limit- name: rate_limit scope: action limit: action: llm_call max_per_minute: 20 action: SIGSTOP reason: Rate limit exceeded与前两条不同这条规则作用于action而非output它对llm_call这类内核动作按分钟限流每分钟最多 20 次超限触发SIGSTOP。它说明策略引擎不仅能检查内容也能治理行为频率与 Agent OS 中rate_limiter等模块的思路一致。审计配置 auditaudit: enabled: true log_path: ./logs/chat-audit.log include: - user_messages - agent_responses - policy_checks - signals开启审计后用户消息、Agent 回复、策略检查结果与信号事件都会被写入./logs/chat-audit.log。这呼应了 Agent OS 的飞行记录器flight recorder理念——README 中展示了kernel.flight_recorder.query(agent_id..., limit10)的回查方式agent-os/README.md每次内核动作自动留痕。架构用户空间与内核空间的边界原文档用一张 ASCII 架构图说明分层关系┌─────────────────────────────────────────────┐ │ CHAT INTERFACE │ │ User input → Agent → Streamed response │ ├─────────────────────────────────────────────┤ │ KERNEL SPACE │ │ Policy Engine ◄─► Memory (EMK) │ │ - Content filter - Conversation history │ │ - Output validation - Context management │ └─────────────────────────────────────────────┘对照源码这张图可以拆成两个可验证的层面CHAT INTERFACE用户空间interactive_chat()中的input()/print()循环、chat_agent里的 OpenAI 调用全部运行在应用进程内KERNEL SPACE内核空间KernelSpace拦截kernel.execute调用先经PolicyEngine做内容过滤与输出校验同时与EpisodicMemory交互完成历史读取与写入。需要明确的是这里的内核空间是应用层治理中间件而非操作系统内核隔离。Agent OS 官方文档对此有专门说明This is application-level enforcement (Python middleware), not OS kernel-level isolation. Agents run in the same process. For true isolation, run agents in containers.agent-os/README.md。理解这一点有助于正确评估它的能力边界策略强制是确定性的not by hoping the LLM follows instructions见同文档 L262但进程级沙箱仍需容器等外部手段。流式输出与治理的协同原文档 Features 一节提到 Responses stream in real-time while being policy-checked回复边流式输出边被策略检查。虽然当前示例文件中的chat_agent是整段生成后统一检查但这一设计方向与 Agent OS 的治理哲学一致检查发生在返回给用户之前而不是事后补救。对于需要逐 token 流式的生产实现你可以将kernel.execute的检查点前移到每个流式分片命中SIGSTOP时中断流并转人工审查——策略文件的结构scope: output 正则 信号动作无需任何改动。复用与扩展建议基于以上分析把这个示例改造成自己的受治理聊天机器人只需三步复制策略骨架保留policies.yaml的kernel/signals/policies/audit四段结构替换deny.patterns为你业务域的正则例如金融场景加no_pii的卡号、身份证模式内容社区加广告或仇恨言论模式按严重度分配信号可人工复核的用SIGSTOP零容忍的用SIGKILL并同步在聊天循环里处理两类异常示例已演示SIGSTOP/SIGKILL两种分支把记忆换成完整 EMKmemory.py文件头明确建议生产环境使用完整 EMK 模块从agent_os.emk导入EpisodicMemory官方 README 的 Governed chatbot 示例即用from agent_os.emk import EpisodicMemory见 agent-os/README.md以获得真正的摘要压缩与长程记忆能力。小结chat-agent 示例用约两百行代码演示了 Agent OS 治理模型的最小闭环KernelSpace提供确定性的策略强制与信号机制EpisodicMemory提供可裁剪的会话记忆policies.yaml以内容过滤 PII 拦截 调用限流三条规则覆盖了内容安全、数据泄露与资源滥用三类常见治理诉求。读懂这个示例你就掌握了 Agent OS 面向聊天场景的完整接线范式——记忆、策略、信号、审计四要素齐备可以直接平移到 SQL Agent、客服机器人等更多场景官方仓库examples/目录还提供了 20 可运行示例见 agent-os/examples。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考