1. 项目概述Nomos一个为安全智能体世界而生的框架最近在探索智能体Agent应用开发时我一直在寻找一个能兼顾灵活性、安全性和可观测性的框架。市面上的选择不少但要么过于学术化难以落地要么过于简单缺乏对复杂工作流和潜在风险的管控。直到我遇到了safe-agentic-world/nomos这个项目它精准地切中了我的痛点如何在一个可控、可审计、可复现的环境里构建和运行由多个智能体协作的复杂系统。简单来说Nomos 是一个用于构建和运行“多智能体系统”Multi-Agent System, MAS的框架。它不只是一个简单的任务编排器更像是一个为智能体打造的“沙盒世界”。在这个世界里每个智能体可以是LLM驱动的也可以是规则引擎都有自己的角色、能力和目标它们通过结构化的消息进行通信和协作共同完成一个更大的任务。而框架本身则提供了环境模拟、状态管理、安全护栏、以及完整的执行日志和可观测性工具。这解决了什么实际问题呢想象一下你要开发一个自动化客户支持系统。一个智能体负责理解用户意图一个负责查询知识库另一个负责生成友好且合规的回复可能还需要一个“审核员”智能体来检查回复内容是否恰当。如果这些智能体直接、无约束地调用LLM API和数据库你会面临调试困难、成本不可控、输出不稳定、甚至产生有害内容的风险。Nomos 通过其“世界”World模型为这些交互提供了一个清晰、边界明确的舞台让开发、调试和运维都变得有章可循。无论你是想研究多智能体协作的学者还是希望将智能体技术应用于实际业务场景的工程师Nomos 都提供了一个坚实且深思熟虑的起点。它尤其适合那些对系统可靠性、行为可解释性和运行安全性有较高要求的项目。2. 核心设计理念与架构拆解2.1 从“智能体编程”到“世界模拟”的范式转变很多初代智能体框架的思维是“任务链”或“工作流”将一系列对LLM的调用串联起来。Nomos 的核心理念则更进一步它引入了“世界”World作为一等公民。你可以把这个世界理解为一个虚拟的、离散事件驱动的模拟环境。在这个世界里存在几个核心实体世界World整个系统的容器和运行时。它维护着全局状态State管理着所有实体智能体、对象并驱动着模拟时钟Tick前进。智能体Agent世界的参与者。每个智能体封装了感知Perception、决策Policy和行动Action的能力。感知函数决定它能从世界状态中观察到什么策略函数通常由LLM驱动基于观察决定做什么行动函数则将决策转化为对世界状态的修改。对象Object世界中可以被感知和操作的事物。例如在一个电商场景中商品、订单、购物车都可以是对象。消息Message智能体间通信的基本单元。消息有发送者、接收者、内容和类型所有通信都通过世界进行路由这天然提供了通信日志和拦截点。这种设计带来了几个关键优势状态集中化所有变化都通过修改世界状态发生使得系统在任何时刻的状态都是明确且可序列化的极大方便了调试、回滚和复现。自然的并发与顺序控制世界以“回合”Tick为单位推进。在每个回合所有智能体并行地进行“感知-决策”但“行动”的执行顺序可以由世界调度器决定这避免了资源竞争和状态混乱。安全边界清晰智能体不能直接访问外部API或数据库它们只能通过世界提供的“动作”来影响环境。框架可以在动作执行前后插入检查Guardrails例如检查输入/输出是否合规、调用频率是否过高等。2.2 核心组件深度解析Nomos 的架构可以分解为以下几个层次理解它们之间的关系对高效使用框架至关重要。2.2.1 状态State与观察Observation机制世界状态是一个嵌套的数据结构通常是一个字典。智能体不能直接读取完整状态而是通过其perceive方法接收一个“观察”。这是一个关键的安全和抽象设计。例如一个智能体可能只被允许观察到与它相关的对话历史而无法看到其他智能体的私有数据。# 示例一个智能体的感知函数 def perceive(self, world_state): # 从全局world_state中提取该智能体可见的部分 my_observation { conversation_history: world_state[conversations].get(self.id, []), available_actions: world_state[available_actions][self.role] } return my_observation注意设计良好的观察空间是构建高效智能体的第一步。观察内容过多会增加LLM的上下文负担和成本过少则可能导致智能体因信息不足而做出错误决策。通常需要根据智能体的角色进行精心裁剪。2.2.2 策略Policy与动作Action执行智能体的核心是其策略函数policy它接收观察并输出一个动作Action。动作是一个定义了如何修改世界状态的数据结构。# 示例一个基于LLM的决策策略 async def policy(self, observation): prompt self._build_prompt(observation) llm_response await self.llm_client.chat(prompt) # 解析LLM响应将其转化为一个标准的Action对象 action self._parse_response_to_action(llm_response) return action动作被提交给世界后由世界的execute_action方法处理。这里是执行副作用如调用API、更新状态和触发后续流程的地方。2.2.3 回合Tick调度与事件循环Nomos 世界的运行基于一个主循环。每一轮循环称为一个“Tick”包含以下阶段感知阶段为每个智能体调用perceive生成各自的观察。决策阶段为每个智能体异步调用policy生成动作。行动阶段按照预定顺序如根据优先级、依赖关系执行收集到的动作。每个动作的执行会更新世界状态。事件触发状态变化可能触发预定义的事件如“当订单状态变为已支付时”进而激活其他智能体或流程。这种离散的、回合制的推进方式使得整个系统的行为变得确定且可调试。你可以轻松地记录下第N个Tick的世界状态并在任何时候从那个状态重新开始模拟。3. 从零开始构建一个Nomos多智能体应用理论讲得再多不如动手实践。让我们以一个简化的“智能内容创作工坊”为例构建一个包含三个智能体的系统策划Agent、写作Agent和评审Agent。3.1 环境搭建与项目初始化首先确保你的Python环境在3.9以上。创建项目并安装Nomos。# 创建项目目录 mkdir content-workshop cd content-workshop python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 安装Nomos。由于它可能处于活跃开发中建议从GitHub安装 pip install safe-agentic-world[nomos] githttps://github.com/safe-agentic-world/nomos.git # 同时安装你需要的LLM SDK例如OpenAI pip install openai项目结构规划如下content-workshop/ ├── agents/ │ ├── __init__.py │ ├── planner.py # 策划Agent │ ├── writer.py # 写作Agent │ └── reviewer.py # 评审Agent ├── objects/ │ ├── __init__.py │ └── content_brief.py # 内容简报对象 ├── world.py # 自定义World类 ├── config.py # 配置API密钥等 └── main.py # 应用入口3.2 定义世界状态与核心对象世界状态是我们系统的单一数据源。我们先定义核心的ContentBrief对象和初始世界状态。# objects/content_brief.py from typing import TypedDict, List, Optional class ContentBrief(TypedDict): 内容简报作为世界中的一个核心对象 topic: str target_audience: str key_points: List[str] tone: str outline: Optional[List[str]] # 由策划Agent生成 draft: Optional[str] # 由写作Agent生成 feedback: Optional[str] # 由评审Agent生成 status: str # 例如”initialized“, ”outlined“, ”drafted“, ”reviewed“, ”approved“# world.py from nomos import World from typing import Dict, Any from .objects.content_brief import ContentBrief class ContentWorkshopWorld(World): def __init__(self): super().__init__() # 初始化世界状态 self.state: Dict[str, Any] { content_briefs: {}, # key: brief_id, value: ContentBrief active_brief_id: None, message_board: [], # 用于智能体间通信的公共留言板 workflow_stage: idle, # 追踪整体工作流阶段 } def create_new_brief(self, topic: str, audience: str, points: List[str], tone: str) - str: 创建一个新的内容简报对象并放入世界状态 import uuid brief_id str(uuid.uuid4())[:8] new_brief: ContentBrief { topic: topic, target_audience: audience, key_points: points, tone: tone, outline: None, draft: None, feedback: None, status: initialized } self.state[content_briefs][brief_id] new_brief self.state[active_brief_id] brief_id self.state[workflow_stage] planning return brief_id3.3 实现三个核心智能体3.3.1 策划Agent (PlannerAgent)它的职责是根据初始简报生成一份详细的内容大纲。# agents/planner.py from nomos import Agent from typing import Dict, Any import openai import asyncio class PlannerAgent(Agent): def __init__(self, agent_id: str, openai_api_key: str): super().__init__(agent_id) self.client openai.AsyncOpenAI(api_keyopenai_api_key) self.role 内容策划专家 def perceive(self, world_state: Dict[str, Any]) - Dict[str, Any]: 只观察当前活跃的简报和需要策划的任务 active_id world_state.get(active_brief_id) if not active_id: return {task: wait} brief world_state[content_briefs][active_id] # 只有当简报处于初始化状态时策划Agent才工作 if brief[status] ! initialized: return {task: wait} return { task: generate_outline, brief: {k: v for k, v in brief.items() if k not in [outline, draft, feedback, status]} } async def policy(self, observation: Dict[str, Any]) - Dict[str, Any]: 基于观察调用LLM生成大纲 if observation[task] wait: return {action: no_op} # 无操作动作 brief observation[brief] prompt f 你是一位{self.role}。请根据以下要求创作一份详细的内容大纲。 主题{brief[topic]} 目标读者{brief[target_audience]} 核心要点{, .join(brief[key_points])} 文风{brief[tone]} 请输出一个结构化的大纲包含引言、主体至少3个部分每部分有子要点和结论。 直接输出大纲内容不要额外解释。 try: response await self.client.chat.completions.create( modelgpt-4-turbo-preview, messages[{role: user, content: prompt}], temperature0.7, max_tokens800 ) outline response.choices[0].message.content.strip() # 返回一个更新世界状态的动作 return { action: update_brief, field: outline, value: outline, next_status: outlined } except Exception as e: # 错误处理返回一个记录错误的动作 return { action: post_message, message: fPlanner出错: {str(e)}, level: error }3.3.2 写作Agent (WriterAgent) 与 评审Agent (ReviewerAgent)写作Agent的perceive会检查是否有状态为”outlined“且无草稿的简报然后根据大纲和简报撰写草稿。评审Agent则检查是否有状态为”drafted“且无反馈的简报然后阅读草稿并给出修改意见。它们的policy函数结构与策划Agent类似但提示词Prompt和触发的动作不同。实操心得智能体职责的单一性与提示词设计每个智能体的perceive方法应使其职责尽可能单一。例如写作Agent不应该去关心评审逻辑。这能降低每个智能体的认知负担使其提示词更专注、效果更好。提示词中应清晰包含其角色、当前任务上下文从观察中获得以及具体的输出格式指令。好的格式指令能极大简化后续动作的解析。3.4 编织世界连接智能体与执行逻辑我们需要在世界中注册智能体并实现动作的执行逻辑。# world.py (续) class ContentWorkshopWorld(ContentWorkshopWorld): async def execute_action(self, agent_id: str, action: Dict[str, Any]) - Dict[str, Any]: 执行智能体提交的动作更新世界状态 action_type action.get(action) active_id self.state[active_brief_id] if action_type no_op: return {result: skipped} elif action_type update_brief and active_id: field action[field] value action[value] self.state[content_briefs][active_id][field] value if next_status in action: self.state[content_briefs][active_id][status] action[next_status] self.state[workflow_stage] action[next_status] return {result: success, updated_field: field} elif action_type post_message: self.state[message_board].append({ from: agent_id, message: action[message], level: action.get(level, info) }) return {result: message_posted} else: # 处理未知动作 self.state[message_board].append({ from: system, message: f收到未知动作来自 {agent_id}: {action_type}, level: warning }) return {result: unknown_action} async def run_workshop(self, initial_brief: Dict): 运行一次完整的内容工坊流程 # 1. 创建初始简报 brief_id self.create_new_brief(**initial_brief) print(f开始处理简报: {brief_id}) # 2. 实例化智能体 (实际项目中应从配置加载) from .agents import PlannerAgent, WriterAgent, ReviewerAgent import os planner PlannerAgent(planner_1, os.getenv(OPENAI_API_KEY)) writer WriterAgent(writer_1, os.getenv(OPENAI_API_KEY)) reviewer ReviewerAgent(reviewer_1, os.getenv(OPENAI_API_KEY)) agents [planner, writer, reviewer] # 3. 定义简单的回合逻辑运行直到简报状态变为 reviewed 或达到最大回合数 max_ticks 10 for tick in range(max_ticks): print(f\n--- Tick {tick} ---) current_stage self.state[workflow_stage] print(f当前阶段: {current_stage}) # 感知阶段 observations {} for agent in agents: obs agent.perceive(self.state) observations[agent.agent_id] obs # 决策阶段 (异步并行) tasks [agent.policy(obs) for agent, obs in zip(agents, observations.values())] actions await asyncio.gather(*tasks, return_exceptionsTrue) # 行动阶段 (顺序执行) for agent, action in zip(agents, actions): if isinstance(action, Exception): print(fAgent {agent.agent_id} 决策出错: {action}) continue result await self.execute_action(agent.agent_id, action) print(fAgent {agent.agent_id} 执行 {action.get(action)}: {result}) # 检查终止条件 active_brief self.state[content_briefs][self.state[active_brief_id]] if active_brief[status] reviewed: print(f\n流程完成于 Tick {tick}!) print(f最终草稿长度: {len(active_brief.get(draft, ))} 字符) print(f评审反馈: {active_brief.get(feedback, )[:200]}...) break # 模拟一点延迟便于观察 await asyncio.sleep(0.5) else: print(f达到最大回合数 {max_ticks}流程未自然结束。) return self.state3.5 运行与观察最后创建一个主入口文件来启动整个世界。# main.py import asyncio import os from dotenv import load_dotenv from world import ContentWorkshopWorld load_dotenv() # 从.env文件加载OPENAI_API_KEY async def main(): workshop ContentWorkshopWorld() initial_brief { topic: 如何利用Nomos框架构建可靠的多智能体应用, audience: 有一定Python和LLM基础的中级开发者, points: [Nomos的核心概念世界、智能体、状态, 与普通任务链框架的区别, 安全性和可观测性设计, 实战构建步骤], tone: 技术博客清晰、务实、附带代码示例 } final_state await workshop.run_workshop(initial_brief) # 输出最终成果 active_brief final_state[content_briefs][final_state[active_brief_id]] print(\n *50) print(最终内容简报:) for key, value in active_brief.items(): if key in [outline, draft] and value: print(f\n{key.upper()}:\n{value[:500]}...) # 只打印前500字符 elif key feedback and value: print(f\n{key.upper()}:\n{value}) if __name__ __main__: asyncio.run(main())运行python main.py你将看到一个由三个智能体协作依次完成策划、写作、评审的完整流程在控制台中逐步上演。每个回合的状态变化、智能体的动作和执行结果都清晰可见。4. 高级特性与生产级考量基础流程跑通后我们需要考虑如何将Nomos应用到更复杂、更可靠的生产环境中。4.1 安全护栏Guardrails与动作验证在智能体动作真正执行前进行校验是保障系统安全的关键。Nomos允许你在execute_action中或之前插入校验逻辑。# 在execute_action方法内部或之前添加 def _validate_action(self, agent_id: str, action: Dict) - Tuple[bool, str]: 验证动作的合法性 # 1. 权限检查该agent是否被允许执行此类动作 allowed_actions self.agent_permissions.get(agent_id, []) if action.get(action) not in allowed_actions: return False, fAgent {agent_id} 无权执行 {action[action]} # 2. 输入检查防止Prompt注入或恶意输入 if action.get(action) update_brief: value action.get(value, ) if len(value) 10000: # 限制输入长度 return False, 输入内容过长 # 可以添加敏感词过滤等 if 恶意关键词 in value: return False, 输入包含违规内容 # 3. 状态一致性检查当前世界状态是否允许此动作 if action.get(action) update_brief and action.get(field) draft: # 只有大纲已完成后才能写草稿 brief self.state[content_briefs][self.state[active_brief_id]] if brief[status] ! outlined: return False, 当前状态不允许撰写草稿 return True, # 在execute_action中调用 is_valid, reason self._validate_action(agent_id, action) if not is_valid: self.state[message_board].append({ from: system_guardrail, message: f动作被拦截: {reason}, level: error }) return {result: rejected, reason: reason}4.2 可观测性Observability与调试Nomos的回合制设计和集中状态管理天生适合做深度观测。状态快照在每个Tick结束后将self.state序列化如JSON并存储或发送到监控系统。这让你可以随时“时光倒流”查看任意时刻系统的完整情况。动作与观察日志记录每个智能体每回合的观察输入和动作输出。这对于分析智能体决策逻辑、优化Prompt至关重要。性能指标在execute_action中记录每个动作的执行耗时、LLM调用的Token消耗等便于进行成本分析和性能优化。# 一个简单的日志装饰器示例 def log_tick(func): async def wrapper(world, *args, **kwargs): tick_start time.time() result await func(world, *args, **kwargs) tick_duration time.time() - tick_start log_entry { tick: world.current_tick, state_snapshot: json.dumps(world.state, defaultstr, ensure_asciiFalse), duration: tick_duration, timestamp: datetime.now().isoformat() } # 写入文件或发送到日志服务 with open(frun_log_{world.run_id}.jsonl, a) as f: f.write(json.dumps(log_entry) \n) return result return wrapper # 装饰run_workshop方法 class ContentWorkshopWorld(ContentWorkshopWorld): log_tick async def run_workshop(self, initial_brief: Dict): # ... 原有逻辑4.3 扩展性与自定义调度默认的回合制是并行感知、顺序执行动作。对于更复杂的依赖关系你可能需要自定义调度器。基于事件的调度除了主循环可以让状态变更触发事件。例如当brief[status]变为”outlined“时自动唤醒WriterAgent而不是让它每回合都检查。优先级队列为动作引入优先级字段在execute_action阶段按优先级排序执行。子世界与层次化复杂的系统可以设计成层次化的世界。一个“主世界”管理宏观流程和协调每个智能体内部又可以是一个“子世界”管理其私有的复杂推理步骤。Nomos的架构对此有良好的支持潜力。5. 常见陷阱、问题排查与优化策略在实际使用Nomos构建应用时你可能会遇到以下典型问题。5.1 智能体陷入循环或停滞现象流程运行几个Tick后状态不再变化智能体反复输出”no_op“或重复动作。排查思路检查观察空间首先打印每个智能体每回合的observation。很可能某个智能体的perceive逻辑有误导致它永远看不到触发其行动的状态条件。例如WriterAgent的perceive可能错误地判断brief[”status“] ! ”outlined“导致它永远认为没活干。检查动作执行结果在execute_action中打印详细的执行结果和更新后的状态。确认动作确实按预期修改了世界状态。常见错误是在更新状态时写错了字典的Key。检查终止条件确认你的循环终止条件如status ”reviewed“能被正确触发。可能是评审Agent的policy没有正确设置next_status。避坑技巧在开发初期大量使用print日志或在世界状态中设置一个debug_log列表来记录关键节点的数据。Nomos的状态可序列化特性使得你可以轻松地将整个世界的状态保存下来离线分析。5.2 LLM调用不稳定或成本过高现象响应时好时坏或单次运行消耗大量Token。优化策略精简Prompt与观察传递给LLM的观察内容要尽可能精简。只包含该智能体做决策必需的信息。避免将整个庞大的世界状态都塞进去。结构化输出与解析要求LLM以严格的JSON或特定格式输出并在policy方法中实现健壮的解析逻辑包含重试和降级处理。使用像Pydantic这样的库来定义输出模型并利用LLM的function calling或JSON mode特性可以极大提高稳定性。缓存与记忆对于相同的观察输入决策结果可能相同。可以考虑为智能体添加一个简单的缓存基于观察内容的哈希值避免重复调用LLM。对于需要历史上下文的情况在观察中提供精炼的摘要而非完整的对话历史。模型选型非核心推理步骤可以使用更便宜、更快的模型如gpt-3.5-turbo核心创意生成步骤再用更强大的模型如gpt-4。5.3 系统难以调试与复现现象线上出现问题但无法还原当时的场景进行调试。解决之道充分利用Nomos的确定性优势。完整状态序列化如前所述在每个Tick后记录完整状态快照。这相当于拥有了一个“黑匣子”。种子与随机性控制如果智能体的策略中涉及随机性如LLM的temperature确保记录下随机种子。在复现时使用相同的种子。录制与回放设计一个“录制模式”将一次成功运行的完整轨迹包括所有LLM的请求和响应保存下来。然后实现一个“回放模式”在回放时智能体的policy不再真实调用LLM而是从录制的轨迹中读取预先确定的响应。这能实现百分百的确定性和离线调试。5.4 多智能体协作效率低下现象智能体间沟通不畅需要很多回合才能达成一致或者做出矛盾决策。优化方向设计更高效的通信协议不要只依赖原始文本消息。可以定义结构化的通信动作如ProposePlan、Vote、RequestClarification等并让智能体学会使用这些协议。引入协调者或管理者智能体增加一个专用的CoordinatorAgent它的职责是观察全局状态分配任务解决冲突。它可以通过向其他智能体发送特定的指令动作如AssignTask来指导工作流。共享黑板Blackboard模式除了定向消息强化世界状态中“共享黑板”如我们例子中的message_board的设计。智能体可以将中间结论、投票结果、待决议题发布到黑板上供其他智能体读取这比一对一的通信更高效。构建基于Nomos的多智能体系统是一个在“控制”与“自主性”之间寻找平衡的艺术。开始时建议给予智能体较窄的观察范围和明确的动作集让系统先可靠地跑起来。随着你对系统行为信心的增加再逐步扩大智能体的自主权。始终记住那个清晰、可审计、可回滚的世界状态是你应对复杂性的最强有力工具。