从散装脚本到智能体操作系统：AgentOS架构设计与工程实践

2026/8/4 8:17:46

1. 从“散装”脚本到“操作系统”：为什么我们需要AgentOS

如果你和我一样，在AI Agent这个领域折腾过一阵子，大概率会经历这样一个阶段：手头攒了一堆Python脚本，每个脚本负责一个特定任务，比如一个用来调用大模型API，一个用来处理文件，还有一个用来发邮件或者调用某个Webhook。一开始，项目简单，几个脚本互相调用一下，main.py里写点逻辑，也能跑起来。但随着想法越来越多，你想让这些“智能体”能记住对话历史，能根据上下文选择不同的工具，甚至能自主规划一系列任务时，代码就开始变得一团糟。全局变量满天飞，配置文件散落在各处，添加一个新功能就像是在一堆意大利面条里再塞进一根，调试起来更是噩梦。

这就是我决定停下来，重新思考如何组织代码的契机。我不需要另一个庞大、复杂、学习曲线陡峭的“框架”，我需要的是一个坚实、清晰、可扩展的基底——一个属于我自己的“智能体操作系统”雏形。今天要分享的，就是如何用最朴素的思想——一套精心设计的文件夹结构和一个核心的循环逻辑——来构建这个基底，我称之为“AgentOS”。它不是一个可以直接pip install的库，而是一种工程实践模式，能让你从“写脚本”进化到“构建系统”，从容应对日益复杂的Agent需求。

这套方法的核心价值在于“分离关注点”和“流程标准化”。把大脑（LLM调用）、记忆（状态管理）、工具（能力扩展）、决策（工作流）这些模块清晰地拆分开，然后用一个主循环把它们像齿轮一样咬合起来。你会发现，之后无论你想实验新的记忆方式、接入新的模型API，还是设计复杂的多步推理链，都像是在一个结构清晰的工厂里更换或添加生产线模块，而不是在混乱的车间里手忙脚乱。

2. AgentOS核心架构：文件夹即蓝图

我们先抛开代码，看看最终我希望项目文件夹长什么样。这个结构本身就是系统设计的直观体现：

your_agentos_project/ ├── core/ │ ├── __init__.py │ ├── agent_loop.py # 核心循环引擎 │ ├── state_manager.py # 状态与记忆管理中心 │ └── router.py # 意图识别与路由器 ├── brains/ │ ├── __init__.py │ ├── openai_brain.py # 基于OpenAI API的“大脑” │ ├── claude_brain.py # 基于Anthropic Claude的“大脑” │ └── brain_base.py # 所有“大脑”的抽象基类 ├── tools/ │ ├── __init__.py │ ├── calculator.py # 计算器工具 │ ├── web_searcher.py # 网络搜索工具 │ ├── file_ops.py # 文件操作工具 │ └── tool_base.py # 所有工具的抽象基类 ├── skills/ │ ├── __init__.py │ ├── email_summarizer.py # 邮件总结技能 │ ├── data_analyzer.py # 数据分析技能 │ └── skill_base.py # 所有技能的抽象基类 ├── memory/ │ ├── __init__.py │ ├── buffer_memory.py # 简易对话缓冲区记忆 │ ├── vector_memory.py # 基于向量数据库的记忆 │ └── memory_base.py # 记忆系统的抽象基类 ├── config/ │ └── settings.yaml # 或 .env， 统一配置文件 ├── logs/ │ └── agent_20231027.log # 运行日志目录 ├── tests/ # 单元测试 ├── main.py # 系统启动入口 └── requirements.txt # 依赖清单

现在，我们来逐一拆解每个文件夹的职责和设计逻辑：

core/：系统的中枢神经这里是整个AgentOS的指挥中心。agent_loop.py定义了智能体从感知到行动再到学习的基本工作周期，它是一个可插拔的循环体。state_manager.py是全局状态管家，负责维护当前会话的上下文、用户数据、以及智能体的内部状态（如是否正在执行多步任务）。router.py则像一个调度员，它解析用户输入或当前状态，决定下一步是调用某个tool，还是激活某个skill，或是直接交给brain去生成回复。将它们放在core，意味着它们是系统运行不可或缺的、最稳定的部分。

brains/：模型的抽象与适配层“大脑”是对大语言模型的封装。这里的关键是brain_base.py中定义的基类，它规定了所有大脑必须实现的方法，比如generate(prompt: str) -> str。openai_brain.py和claude_brain.py是具体实现。这样做的好处是，无论底层换用GPT-4、Claude-3还是国产大模型，你只需要实现一个新的XxxBrain类，系统其他部分几乎无需改动。这符合“依赖倒置”原则，高层模块（core）不依赖低层模块（具体LLM），二者都依赖抽象（BaseBrain）。

tools/与skills/：能力的模块化扩展这是最容易混淆的两个概念，我的区分原则是：Tool（工具）是原子操作，Skill（技能）是组合拳。

工具：功能单一、无状态、即用即走。例如calculator.py里的evaluate_expression(expr)函数，给它一个算式“2+2”，它返回“4”。它不关心对话历史，也不进行复杂规划。网络搜索、获取天气、读写特定格式文件，都属于工具。
技能：包含一定逻辑、可能涉及多个工具调用、甚至内部有状态的小型工作流。例如email_summarizer.py这个技能，它可能需要先调用file_ops工具读取邮件文件，然后调用brain进行总结，最后可能再调用某个格式化工具整理输出。技能更像是一个有明确目标的“小程序”。

将二者分离，保持了系统的清晰度。简单任务用工具快速解决，复杂任务用技能封装复用。

memory/：赋予智能体“记忆”记忆系统是智能体体现“智能”和“连续性”的关键。buffer_memory.py可能只是维护一个最近N轮对话的列表，简单高效。而vector_memory.py则更高级，它将历史对话通过嵌入模型向量化后存入ChromaDB或Pinecone，实现基于语义的长期记忆检索。记忆基类定义了add(message)和get_relevant_context(query)等接口。在核心循环中，每次行动前，我们都会从记忆系统中获取相关的历史上下文，拼接到给大脑的提示词中。

config/,logs/,tests/：工程化的基石使用统一的settings.yaml管理API密钥、模型参数、开关配置，避免硬编码。独立的logs/目录便于问题追踪和效果分析。而tests/文件夹则是保证每个模块在迭代中依然能正确工作的安全网。这些看似辅助的部分，是项目能否从个人实验走向可靠应用的关键。

3. 心脏：Agent Loop 的详细设计与实现

有了清晰的结构，我们需要一个动力核心将它们驱动起来，这就是Agent Loop。它不是一个简单的while True循环，而是一个定义了明确阶段的状态机。下面是我在core/agent_loop.py中实现的一个经典循环版本：

# core/agent_loop.py import logging from typing import Optional, Any from .state_manager import StateManager from .router import Router from brains.brain_base import BaseBrain from memory.memory_base import BaseMemory class AgentLoop: def __init__(self, brain: BaseBrain, memory: BaseMemory, router: Router, initial_state: Optional[dict] = None): self.brain = brain self.memory = memory self.router = router self.state_manager = StateManager(initial_state or {}) self.logger = logging.getLogger(__name__) def run_for_n_iterations(self, user_input: str, max_iterations: int = 10): """运行主循环，处理一次用户输入，可能触发多轮内部迭代。""" # 迭代0：初始化，将用户输入存入记忆和状态 self.memory.add({"role": "user", "content": user_input}) self.state_manager.update({"current_goal": user_input, "iteration": 0}) final_response = None for i in range(max_iterations): self.logger.info(f"--- 迭代开始 [第 {i+1} 轮] ---") self.state_manager.update({"iteration": i+1}) # 阶段1：感知与规划 - 决定下一步做什么 plan = self._plan_next_action() if plan.get("action") == "respond_directly": # 大脑认为可以直接回复了 final_response = plan.get("response") break elif plan.get("action") == "use_tool": # 阶段2：执行 - 调用工具 tool_result = self._execute_tool(plan) # 将工具执行结果作为系统消息存入记忆，供下一轮参考 self.memory.add({"role": "system", "content": f"工具执行结果: {tool_result}"}) self.state_manager.update({"last_tool_result": tool_result}) # 继续循环 elif plan.get("action") == "run_skill": # 阶段2：执行 - 运行技能（技能内部可能包含自己的小循环） skill_output = self._execute_skill(plan) self.memory.add({"role": "system", "content": f"技能执行完成: {skill_output}"}) self.state_manager.update({"last_skill_output": skill_output}) else: self.logger.error(f"未知的规划动作: {plan}") final_response = "系统内部规划出错。" break # 安全阀：防止无限循环 if i == max_iterations - 1: final_response = "已达到最大思考步数，未能得出最终结论。" self.logger.warning("循环因达到最大迭代次数而终止。") # 循环结束，返回最终响应 if final_response: self.memory.add({"role": "assistant", "content": final_response}) return final_response else: error_msg = "循环意外结束，无响应。" self.logger.error(error_msg) return error_msg def _plan_next_action(self) -> dict: """核心规划器：结合记忆、状态和大脑，决定下一步动作。""" # 1. 从记忆系统中获取与当前目标相关的历史上下文 current_goal = self.state_manager.get("current_goal") relevant_history = self.memory.get_relevant_context(current_goal, k=5) # 2. 构建规划提示词 planning_prompt = f""" 你是一个任务规划器。当前用户目标是：{current_goal} 相关的历史对话和系统记录如下： {relevant_history} 当前系统状态：{self.state_manager.get_state_snapshot()} 请分析，为了达成用户目标，下一步应该做什么？请从以下选项中选择，并按要求格式回复： A. RESPOND_DIRECTLY - 如果已有足够信息可以直接回答用户。 B. USE_TOOL - 如果需要使用一个工具（如计算、搜索）来获取信息。请指定工具名称和输入参数。 C. RUN_SKILL - 如果需要执行一个复杂技能（如总结、分析）。请指定技能名称和输入参数。 你的回复必须是严格的JSON格式，只包含以下字段： {{ "reasoning": "你的简要推理过程", "action": "RESPOND_DIRECTLY | USE_TOOL | RUN_SKILL", "response": "仅当action为RESPOND_DIRECTLY时提供回复内容", "tool_name": "仅当action为USE_TOOL时提供工具名", "tool_params": {{}} // 仅当action为USE_TOOL时提供参数字典 "skill_name": "仅当action为RUN_SKILL时提供技能名", "skill_params": {{}} // 仅当action为RUN_SKILL时提供参数字典 }} """ # 3. 调用大脑进行规划决策 planning_response = self.brain.generate(planning_prompt) # 4. 解析大脑的JSON输出 try: import json plan = json.loads(planning_response) self.logger.debug(f"规划结果: {plan}") return plan except json.JSONDecodeError as e: self.logger.error(f"规划器返回了非JSON内容: {planning_response}") # 降级处理：返回一个安全的后备计划 return {"action": "RESPOND_DIRECTLY", "response": "我在思考时遇到了点问题，请重新表述您的需求。"} def _execute_tool(self, plan: dict) -> Any: """根据规划结果调用具体的工具。""" tool_name = plan.get("tool_name") tool_params = plan.get("tool_params", {}) self.logger.info(f"执行工具: {tool_name}, 参数: {tool_params}") # 这里应该有一个工具注册表或发现机制，简化起见，我们假设通过导入获取 # 实际项目中，可以使用一个中央注册表来管理所有可用工具 from tools import TOOL_REGISTRY if tool_name in TOOL_REGISTRY: tool_class = TOOL_REGISTRY[tool_name] try: result = tool_class().execute(**tool_params) return result except Exception as e: error_msg = f"工具 {tool_name} 执行失败: {str(e)}" self.logger.exception(error_msg) return error_msg else: error_msg = f"未知工具: {tool_name}" self.logger.error(error_msg) return error_msg def _execute_skill(self, plan: dict) -> Any: """根据规划结果调用具体的技能。""" # 实现逻辑与_execute_tool类似，但技能可能更复杂，甚至内部有自己的状态。 # 技能可以看作是一个更高级的、封装的Agent。 skill_name = plan.get("skill_name") skill_params = plan.get("skill_params", {}) self.logger.info(f"执行技能: {skill_name}, 参数: {skill_params}") # ... 类似的查找和调用逻辑 # 技能执行可能会返回一个复杂对象或字符串 return f"技能 '{skill_name}' 执行完成。"

这个循环的设计精髓在于“思考-行动”的迭代。智能体不是一次性生成答案，而是通过多轮“规划-执行-观察结果-再规划”的循环来逼近目标。_plan_next_action方法利用大脑进行元认知，判断当前信息是否充足。如果不充足，它必须明确指定下一步要使用的tool或skill及其参数。这种设计将“决策逻辑”很大程度上交给了LLM，而我们只需要提供清晰的选项和格式约束。

注意：规划提示词（Planning Prompt）的质量直接决定了系统的可靠性和效率。你需要精心设计提示词，让LLM理解选项的含义，并稳定输出可解析的JSON。在实际应用中，可能需要加入少量示例（Few-shot）到提示词中，或者使用LLM的Function Calling特性来获得更稳定的结构化输出。

4. 状态管理与记忆系统的协同设计

在循环中，StateManager和Memory是两个紧密协作但又职责分明的组件。理解它们的区别是设计健壮Agent的关键。

StateManager（状态管理器）管理的是当前会话周期内的、临时的、结构化的运行时数据。它像一个便签本，记录着“现在正在发生什么”。

典型状态数据：current_goal（当前用户目标）、iteration（循环迭代次数）、last_tool_result（上一次工具调用结果）、is_waiting_for_user（是否在等待用户输入）、current_skill_step（如果技能是多步骤的，记录当前步骤）。
特点：生命周期短（通常一次对话或任务），数据结构明确（字典），读写频繁，用于控制流程。

Memory（记忆系统）管理的是跨越会话的、长期的、可供检索的对话历史与知识。它像一个长期档案库。

典型记忆数据：完整的用户与助手对话记录、工具执行的结果日志、从外部获取的重要事实（如搜索到的资料）。
特点：生命周期长，数据量大，需要高效的检索能力（如向量检索），用于提供上下文。

它们的协作流程如下：

用户输入：用户说“帮我总结一下上周项目会议的邮件”。这个输入被同时存入Memory（作为历史记录）和StateManager（作为current_goal）。
规划阶段：_plan_next_action方法从Memory中检索与“总结邮件”相关的历史（比如用户之前提到过“项目A”），并从StateManager获取当前状态（如这是第一轮迭代）。这些信息共同构成规划提示词。
执行与更新：规划决定调用email_summarizer技能。技能执行过程中，StateManager可以记录current_skill: email_summarizer, step: fetching_emails。技能执行成功后，其输出（如“已找到3封相关邮件”）被作为一条系统消息存入Memory，同时StateManager更新last_skill_output。
下一轮迭代：新的循环开始，Memory中现在有了“用户目标”和“已找到邮件”两条记录。规划器基于更丰富的上下文，可能决定下一步是“调用大脑进行总结”。

这种分离带来了灵活性。你可以轻松更换记忆后端（比如从简单的缓冲区切换到向量数据库），而无需改动状态管理逻辑。你也可以在状态中存储一些临时变量（如API调用重试次数），而不会污染长期记忆。

一个实操心得：在StateManager中，我通常会实现一个get_state_snapshot()方法，它返回当前状态的精简、字符串化的版本，方便拼接到给LLM的提示词中。而对于Memory，get_relevant_context(query, k)方法则是核心，它决定了智能体“记得什么”。对于向量记忆，这里的query通常是当前的目标或状态摘要，通过计算余弦相似度找回最相关的k条历史记录。调试时，一定要打印出每次规划时使用的“记忆上下文”和“状态快照”，这是理解智能体决策过程的最重要窗口。

5. 工具与技能的开发规范与注册机制

要让Agent Loop能动态发现和调用tools和skills，一个中央注册表是必不可少的。这避免了在代码中硬编码if tool_name == 'calculator'这样的语句。下面是一个简单的实现范例：

首先，定义好基类，明确契约：

# tools/tool_base.py from abc import ABC, abstractmethod from typing import Any, Dict class BaseTool(ABC): """所有工具的基类。""" name: str = "base_tool" # 工具的唯一标识名 description: str = "工具描述" # 用于告知LLM此工具功能的描述 parameters: Dict[str, Any] = {} # 工具所需的参数JSON Schema @abstractmethod def execute(self, **kwargs) -> Any: """执行工具的核心方法。""" pass # skills/skill_base.py 类似 from abc import ABC, abstractmethod from typing import Any, Dict class BaseSkill(ABC): """所有技能的基类。""" name: str = "base_skill" description: str = "技能描述" parameters: Dict[str, Any] = {} @abstractmethod def run(self, **kwargs) -> Any: """运行技能的核心方法。技能内部可以更复杂，甚至可以调用其他工具或启动子Agent。""" pass

然后，实现具体的工具。工具的实现应该尽可能纯粹和健壮，做好输入验证和错误处理。

# tools/calculator.py import ast import operator from .tool_base import BaseTool class CalculatorTool(BaseTool): name = "calculator" description = "计算一个数学表达式的值。支持加减乘除(+-*/)、乘方(**)和括号。" parameters = { "type": "object", "properties": { "expression": { "type": "string", "description": "数学表达式，例如 '(2 + 3) * 4 ** 0.5'" } }, "required": ["expression"] } def execute(self, expression: str) -> str: """安全地计算数学表达式。禁止使用eval执行任意代码。""" self._validate_expression(expression) try: # 使用ast.literal_eval进行安全评估，但需要先将运算符映射为函数 # 这里简化处理，使用一个受限制的eval替代方案，实际生产环境应用更安全的库如`simpleeval` node = ast.parse(expression, mode='eval') # 可以在这里遍历AST节点，检查只允许数字和运算符 # 为简单演示，我们使用一个安全的评估函数（示例，非生产级） result = self._safe_eval(expression) return str(result) except (SyntaxError, ValueError, TypeError, ZeroDivisionError) as e: return f"计算错误: {type(e).__name__}: {str(e)}" def _validate_expression(self, expr: str): """简单的表达式验证。""" allowed_chars = set("0123456789+-*/.()% **") # 检查是否包含除允许字符外的字母（可能是不安全代码） if any(c.isalpha() and c not in allowed_chars for c in expr): raise ValueError("表达式包含不安全字符") # 其他安全检查... def _safe_eval(self, expr: str): """一个极其简化的安全评估示例。生产环境请使用专业库。""" # 警告：此方法仅为演示，不足以防御所有恶意输入。 # 考虑使用 `simpleeval` 或 `numexpr` 等库。 import re # 移除空格，进行基础替换 expr = expr.replace(' ', '').replace('**', '^') # 将**替换为^以便处理 # 非常基础的验证和计算，实际不可靠 # 这里省略了复杂实现... return "（安全评估逻辑需另行实现）"

最后，创建一个注册机制，让系统启动时能自动发现所有可用的工具和技能。

# tools/__init__.py import pkgutil import importlib from .tool_base import BaseTool TOOL_REGISTRY = {} def register_tool(tool_class): """装饰器，用于注册工具类。""" if not issubclass(tool_class, BaseTool): raise TypeError(f"{tool_class.__name__} 必须继承自 BaseTool") TOOL_REGISTRY[tool_class.name] = tool_class return tool_class def auto_discover_tools(): """自动发现当前目录下所有模块中的工具类并注册。""" package_path = __path__ for _, module_name, _ in pkgutil.iter_modules(package_path): full_module_name = f"{__name__}.{module_name}" try: module = importlib.import_module(full_module_name) for attr_name in dir(module): attr = getattr(module, attr_name) if (isinstance(attr, type) and issubclass(attr, BaseTool) and attr is not BaseTool): # 检查类是否有有效的name属性 if hasattr(attr, 'name') and attr.name: TOOL_REGISTRY[attr.name] = attr except ImportError as e: print(f"导入模块 {module_name} 失败: {e}") continue # 在包初始化时自动发现 auto_discover_tools() # skills/__init__.py 同理，实现SKILL_REGISTRY和auto_discover_skills

这样，当你新增一个tools/weather.py文件并定义一个WeatherTool类后，系统在下次启动时就能自动识别并注册它。在Agent Loop的_execute_tool方法中，就可以直接通过TOOL_REGISTRY[tool_name]来获取并实例化工具了。

开发工具/技能时的注意事项：

输入验证与净化：永远不要信任来自LLM的输入。工具必须对参数进行严格的类型和范围检查，防止注入攻击或意外错误。
错误处理与友好反馈：工具执行失败时，应返回结构化的错误信息（如{"error": true, "message": "..."}），而不仅仅是抛出异常。这能让规划器在下一轮更好地处理故障。
描述清晰：description和parameters的JSON Schema要写得清晰准确，它们是LLM能否正确使用该工具的关键。好的描述就像给LLM的说明书。
保持无状态：工具本身应尽量设计为无状态的函数式服务。状态信息应由上层的StateManager或Memory管理。

6. 从零搭建与运行你的第一个AgentOS实例

理论说了这么多，我们动手搭一个最简单的可运行版本。这个例子将实现一个能进行多轮数学计算的对话智能体。

第一步：项目初始化与依赖创建一个新目录，并建立我们之前讨论的文件夹结构。然后创建requirements.txt：

openai>=1.0.0 python-dotenv

安装依赖：pip install -r requirements.txt。

第二步：实现最简大脑（OpenAI）创建brains/openai_brain.py：

# brains/openai_brain.py import os from openai import OpenAI from .brain_base import BaseBrain class OpenAIBrain(BaseBrain): def __init__(self, model="gpt-3.5-turbo"): self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) self.model = model def generate(self, prompt: str, **kwargs) -> str: try: response = self.client.chat.completions.create( model=self.model, messages=[{"role": "user", "content": prompt}], temperature=0.1, # 规划任务需要低随机性 **kwargs ) return response.choices[0].message.content.strip() except Exception as e: return f"[大脑调用出错] {str(e)}"

第三步：实现简易记忆和状态管理创建memory/buffer_memory.py：

# memory/buffer_memory.py from .memory_base import BaseMemory from typing import List, Dict class BufferMemory(BaseMemory): def __init__(self, max_size=10): self.buffer: List[Dict] = [] self.max_size = max_size def add(self, message: Dict): self.buffer.append(message) if len(self.buffer) > self.max_size: self.buffer.pop(0) # 移除最老的记录 def get_relevant_context(self, query: str, k: int = 5) -> str: # 简易实现：返回最近k条记录 recent = self.buffer[-k:] return "\n".join([f"{m.get('role', 'unknown')}: {m.get('content', '')}" for m in recent]) def clear(self): self.buffer.clear()

创建core/state_manager.py：

# core/state_manager.py class StateManager: def __init__(self, initial_state=None): self.state = initial_state or {} def update(self, new_state: dict): self.state.update(new_state) def get(self, key, default=None): return self.state.get(key, default) def get_state_snapshot(self) -> str: return str(self.state)

第四步：实现一个计算器工具和路由逻辑创建tools/calculator_simple.py（一个更安全的简化版）：

# tools/calculator_simple.py from .tool_base import BaseTool @register_tool # 使用装饰器注册，需在__init__.py中定义 class CalculatorSimpleTool(BaseTool): name = "calculator_simple" description = "进行简单的四则运算（加+ 减- 乘* 除/）。输入应为空格分隔的表达式，如 '3 + 5 * 2'。" parameters = { "type": "object", "properties": { "expression": {"type": "string", "description": "简单的算术表达式，如 '3 + 5 * 2'"} }, "required": ["expression"] } def execute(self, expression: str) -> str: try: # 使用一个更安全的评估方式：这里我们手动解析 # 注意：这个实现仅支持两个数的运算，仅为演示 parts = expression.split() if len(parts) != 3: return "错误：表达式格式应为 '数字 运算符 数字'，例如 '3 + 5'" a, op, b = parts a_num, b_num = float(a), float(b) if op == '+': result = a_num + b_num elif op == '-': result = a_num - b_num elif op == '*': result = a_num * b_num elif op == '/': if b_num == 0: return "错误：除数不能为零" result = a_num / b_num else: return f"错误：不支持的运算符 '{op}'，仅支持 + - * /" return str(result) except ValueError: return "错误：表达式包含非数字字符" except Exception as e: return f"计算过程出错: {str(e)}"

创建core/router.py（一个极简版，实际项目会更复杂）：

# core/router.py class Router: """极简路由器，目前只做简单转发，实际可根据意图识别进行复杂路由。""" def __init__(self): pass def route(self, plan: dict) -> dict: # 在这个简单示例中，我们直接返回规划器的决定。 # 未来可以在这里加入权限检查、负载均衡、技能链选择等逻辑。 return plan

第五步：组装并运行主程序创建main.py：

# main.py import os from dotenv import load_dotenv from core.agent_loop import AgentLoop from core.router import Router from brains.openai_brain import OpenAIBrain from memory.buffer_memory import BufferMemory # 导入tools包以触发自动注册 import tools load_dotenv() def main(): # 1. 初始化核心组件 brain = OpenAIBrain(model="gpt-3.5-turbo") memory = BufferMemory(max_size=6) router = Router() # 2. 创建Agent循环引擎 agent = AgentLoop(brain=brain, memory=memory, router=router) print("简易数学助手Agent已启动。输入'退出'或'quit'结束。") print("-" * 40) while True: try: user_input = input("\n您: ").strip() if user_input.lower() in ['退出', 'quit', 'exit']: print("再见！") break if not user_input: continue # 3. 运行Agent处理本次输入 response = agent.run_for_n_iterations(user_input, max_iterations=5) print(f"助手: {response}") except KeyboardInterrupt: print("\n程序被中断。") break except Exception as e: print(f"系统发生错误: {e}") if __name__ == "__main__": main()

第六步：配置与运行在项目根目录创建.env文件，填入你的OpenAI API密钥：

OPENAI_API_KEY=sk-your-api-key-here

现在，运行python main.py。你可以尝试输入“计算一下3加5乘以2等于多少”。观察控制台日志，你会看到智能体进行规划（可能决定调用计算器工具），执行工具，然后根据结果生成最终回复的过程。

这个实例虽然简单，但完整地演示了AgentOS的核心循环、模块化结构和数据流。你可以在此基础上，轻松地添加新的工具（如网络搜索）、更强大的记忆系统（向量数据库）、或者更复杂的技能，而无需重写主干逻辑。

7. 调试、监控与性能优化实战要点

当你的AgentOS项目逐渐复杂，调试和优化就变得至关重要。以下是我在实际项目中积累的几个关键实践：

1. 结构化日志是生命线不要只用print。为每个核心模块配置独立的logger，并设置不同的日志级别（DEBUG, INFO, WARNING, ERROR）。

# 在core/agent_loop.py的__init__中 import logging self.logger = logging.getLogger(__name__) self.logger.setLevel(logging.DEBUG) # 在规划、执行等关键节点记录详细信息 self.logger.debug(f"规划提示词: {planning_prompt[:200]}...") self.logger.info(f"执行工具: {tool_name}， 参数: {tool_params}") self.logger.error(f"工具执行失败: {error_msg}", exc_info=True)

将日志输出到文件，并配置格式，包含时间戳、模块名、日志级别。这样当Agent行为异常时，你可以像查案一样回溯完整的“思考-行动”链条。

2. 为LLM调用添加护栏（Guardrails）LLM的输出不可控，必须防御性编程。

JSON解析：规划器要求LLM返回JSON，但LLM可能返回非JSON内容。一定要用try...except json.JSONDecodeError包裹，并准备降级方案（如返回一个要求重试的默认规划）。
工具参数验证：即使LLM返回了JSON，其参数也可能不符合工具要求。在工具execute方法内部必须做严格的类型和值验证。
超时与重试：网络调用可能失败。为大脑的generate方法设置超时，并实现简单的重试逻辑（注意指数退避）。
Token限制：记忆上下文可能过长。在memory.get_relevant_context中实现token计数和截断策略，优先保留最相关的信息。

3. 设计可观测性（Observability）除了日志，可以设计一个轻量的“事件总线”或“回调系统”，在关键生命周期节点（如循环开始、规划完成、工具调用前/后）触发事件。这样你可以方便地挂接监控插件，比如：

将每次LLM调用和结果存储到数据库，用于后续分析和提示词优化。
实时在控制台或Web界面可视化Agent的决策过程。
统计工具使用频率和成功率，找出不可靠的工具。

4. 性能优化策略

记忆检索优化：向量检索虽然强大，但每次循环都做全量检索成本高。可以结合缓存：将最近几轮对话的检索结果缓存起来，如果用户问题变化不大，直接使用缓存。
并行工具调用：如果规划器决定同时使用多个不相关的工具（例如，同时查询天气和搜索新闻），可以在_execute_tool中引入异步机制（asyncio）并行执行，缩短回合时间。
规划结果缓存：对于相似的输入和状态，规划结果可能相同。可以计算当前状态和输入的哈希值，缓存规划结果，避免重复调用LLM。但要注意缓存失效条件（如记忆更新了）。

5. 测试策略

单元测试：为每个Tool和Skill编写单元测试，模拟各种正常和异常输入。
集成测试：测试整个AgentLoop，使用Mock对象替代真实的Brain和Tool，验证给定输入和记忆下，循环是否能产生预期的动作序列。
端到端测试：准备一组标准问题（如“北京天气如何？”、“计算123*456”），运行完整Agent，检查最终回复是否符合预期。这类测试运行慢，但能发现模块间交互的深层问题。

构建AgentOS不是一蹴而就的，它是一个迭代过程。从这个小而美的循环和结构开始，每次添加新功能或遇到新问题，都反过来思考如何调整架构来更优雅地适应。这套文件夹结构和循环引擎，就像乐高底板，能让你在上面自由地拼搭出越来越复杂和强大的智能体应用，而不会陷入代码的泥潭。