尧图网站建设 
你好我是资深技术博主。今天我们不聊理论空谈直接围绕“Agent OS”这个热门概念拆解它的核心本质并带你手写一个可运行的轻量级 Agent 运行时框架。先说结论Agent OS 不是一个具体的开源软件也不是某个大厂发布的新操作系统。它是一套面向 AI Agent智能体的设计范式与运行时架构。它的目标是解决一个真实痛点当你的项目里只有一个 LLM 调用时不需要复杂的调度但当你有几十个、上百个 Agent 协同工作需要管理记忆、工具、任务队列、上下文窗口和异常恢复时普通脚本已经无法胜任。这篇文章会从零开始带你理解 Agent OS 的核心概念然后搭建一个具备“Agent 注册、任务调度、工具调用、消息通信、记忆存储”五个核心能力的简化版 Agent OS。整个过程我会尽量用完整的代码和配置示例让你照着敲就能跑起来。如果你是一个刚接触 LLM 应用的开发者本文能帮你建立 Agent 系统设计的整体认知如果你已经在做 Agent 应用本文提供的架构拆分和工程化思路可以直接迁移到你的项目中。1. Agent OS 到底是什么先建立整体认知1.1 从“单 Agent 脚本”到“Agent OS”的演进我们先用一个简单的对比来理解。最原始的 LLM 应用很简单就是一个函数调用from openai import OpenAI client OpenAI() def ask_llm(prompt: str) - str: response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], ) return response.choices[0].message.content print(ask_llm(你好介绍一下你自己))这个脚本一旦跑起来就是一个线性流程用户输入 → 模型推理 → 输出结果。它没有状态没有记忆没有工具没有多 Agent 协作。到第二阶段开发者会在脚本里加入工具调用Function Calling。模型在推理时可能返回一个“调用某个函数”的指令脚本会执行这个函数把结果再传回给模型。tools [ { type: function, function: { name: get_weather, description: 获取指定城市的天气, parameters: { type: object, properties: { city: {type: string} }, required: [city] } } } ]这个阶段还是单 Agent 的脚本化增强。到了第三个阶段当你的业务足够复杂比如需要“一个 Agent 负责数据分析一个 Agent 负责生成报告一个 Agent 负责发送邮件”并且这些 Agent 之间还需要互相传递上下文、共享记忆、按优先级排队执行时你就需要一套能够统一管理它们的运行时环境。这套运行时的角色类似于传统服务器里操作系统的作用传统操作系统Agent OS 对应能力进程管理Agent 生命周期管理内存管理上下文窗口与记忆管理进程间通信IPCAgent 之间的消息传递文件系统长期记忆与知识存储设备驱动工具Tools插件机制调度器任务队列与优先级调度系统日志推理过程监控与追踪这就是 Agent OS 的概念来源。它不是一个具体的软件产品而是一套让 Agent 系统在生产环境中稳定运行所需的基础设施抽象。1.2 Agent OS 解决的四类核心问题通过上面的类比你应该已经理解了 Agent OS 的定位。接下来我们把它要解决的具体问题归纳成四类这四类问题也是后面实战部分的设计依据。第一类会话生命周期管理单个 Agent 请求是无状态的但业务是有状态的。你需要清楚一个任务从进入队列到最终完成经历了哪些阶段一个 Agent 在运行中被中断后如何恢复一个长时任务是否需要暂停和续跑。Agent OS 需要定义清晰的状态机并统一管理这些状态转换。第二类记忆与上下文管理LLM 的上下文窗口是有限的。你不能把所有历史消息都塞进每一次请求。Agent OS 需要设计分级记忆机制比如短期记忆当前会话上下文、长期记忆持久化到向量数据库或数据库、工作记忆当前任务相关的临时数据。第三类工具注册与动态调用Agent 要完成真实世界的任务必须调用外部工具。工具可能是内部函数、REST API、数据库操作等。Agent OS 需要提供统一的工具注册表让 Agent 能够发现工具、了解工具的入参格式并在运行时安全地调用工具。第四类多 Agent 协作与任务分发复杂任务往往需要拆分成多个子任务分配给不同的 Agent 并行或串行执行。Agent OS 需要支持任务拆分、结果汇总、消息传递、冲突协调等能力。1.3 Agent OS 一定需要一个“操作系统”吗这里有一个常见误区并不是所有 Agent 应用都需要一套完整的 Agent OS。对于只有几个 Agent、任务流程固定、不需要长期记忆的小项目用简单的编排框架就足够了。Agent OS 的价值在系统规模变大后才体现出来。这篇文章的实战案例会设计一个“面向中小型项目”的轻量级 Agent OS 框架。虽然叫 OS但它的实现并不复杂核心就是把上面提到的五类能力抽象成五个模块用 FastAPI 加 SQLite 就能支撑一个可用的原型。2. 环境准备与项目初始化在正式写代码之前先来准备环境。由于 Agent OS 本身是一个 Python 后端服务我们需要配置好 Python 环境和相关依赖。2.1 版本说明为了保持通用性这里不依赖某个特定的 LLM 厂商 SDK而是通过统一接口来调用大模型。你可以根据实际情况替换为 OpenAI、通义千问、文心一言或本地部署的 Ollama 模型。示例环境如下操作系统Windows 10 / macOS / Linux 均可Python3.10 或以上版本数据库SQLitePython 内置Web 框架FastAPI消息队列使用 Python 内置的asyncio.Queue做内存队列生产环境可替换为 Redis Stream 或 RabbitMQ2.2 创建项目结构先创建项目目录我建议使用下面的结构清晰且容易扩展agent-os-demo/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── core/ │ │ ├── __init__.py │ │ ├── agent.py # Agent 抽象类 │ │ ├── registry.py # Agent 注册表 │ │ ├── message.py # 消息模型 │ │ ├── scheduler.py # 任务调度器 │ │ ├── memory.py # 记忆存储 │ │ └── tool.py # 工具注册 │ └── agents/ │ ├── __init__.py │ ├── writer_agent.py # 示例 Agent文案写作 │ └── analysis_agent.py # 示例 Agent数据分析 ├── requirements.txt └── README.md2.3 安装依赖创建requirements.txt内容如下fastapi0.111.0 uvicorn0.30.1 openai1.30.0 pydantic2.7.1然后在终端执行pip install -r requirements.txt这里需要注意openai库并不是 Agent OS 的必需依赖但示例代码需要一个稳定的 LLM 调用方式来演示。你可以把它替换成任何你熟悉的大模型 SDK接口不变。3. 核心模块设计与原理拆解在写代码之前我们先花一些时间理解每个核心模块的设计思路。这样后面贴完整代码时你能清楚每一段代码在系统里扮演什么角色。3.1 Agent 抽象Agent 到底是什么在 Agent OS 框架里Agent 不是一个大模型而是“大模型 指令模板 工具权限 上下文策略”的组合体。我定义一个BaseAgent抽象类它具备以下属性nameAgent 的唯一名称用于注册和路由。description描述这个 Agent 擅长什么任务供调度器做任务分配。system_prompt系统提示词定义 Agent 的角色和行为规范。tools该 Agent 可以调用的工具列表。memory该 Agent 使用的记忆对象。每个 Agent 的核心方法是run(task: str, context: dict) - str调度器通过这个方法把任务交给 Agent 执行。# 文件路径app/core/agent.py from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional class BaseAgent(ABC): def __init__( self, name: str, description: str, system_prompt: str, tools: Optional[List[Any]] None, memory: Optional[Any] None, ): self.name name self.description description self.system_prompt system_prompt self.tools tools or [] self.memory memory self.status idle # idle - running - finished self.max_retry 3 abstractmethod async def run(self, task: str, context: Optional[Dict[str, Any]] None) - str: 执行一个具体任务。 参数: task: 任务描述 context: 额外的上下文信息比如用户ID、会话ID 返回: Agent 执行结果字符串 pass async def before_run(self): self.status running async def after_run(self): self.status finished def __repr__(self): return fAgent name{self.name} status{self.status}这里把run定义成异步方法是因为真实的 Agent 任务往往涉及网络调用、工具执行异步能更好地利用系统资源。3.2 注册表Agent 的“电话簿”注册表的作用是让系统知道有哪些 Agent 可用。它提供三个关键能力register注册一个 Agent。get根据名称获取 Agent。list_agents列出所有 Agent 的基本信息。# 文件路径app/core/registry.py from typing import Dict, Optional from .agent import BaseAgent class AgentRegistry: def __init__(self): self._agents: Dict[str, BaseAgent] {} def register(self, agent: BaseAgent) - None: if agent.name in self._agents: raise ValueError(fAgent {agent.name} 已存在请更换名称) self._agents[agent.name] agent def get(self, name: str) - Optional[BaseAgent]: return self._agents.get(name) def list_agents(self) - list: return [ {name: name, description: agent.description} for name, agent in self._agents.items() ]注册表设计模式的要点是“名称唯一”。在大型系统中Agent 名称唯一是路由和调度的基础避免多个同名 Agent 导致任务投递不明确。3.3 消息模型Agent 之间怎么说话在多 Agent 协作中消息模型决定了系统能否扩展。我把消息分成两种AgentMessageAgent 之间的通信消息TaskMessage调度器下发给 Agent 的任务消息# 文件路径app/core/message.py from enum import Enum from typing import Any, Dict, Optional from pydantic import BaseModel, Field from datetime import datetime class MessageType(str, Enum): TASK task RESULT result ERROR error HEARTBEAT heartbeat class Message(BaseModel): msg_id: str Field(default_factorylambda: fmsg_{datetime.now().timestamp()}) type: MessageType sender: str receiver: str content: str metadata: Optional[Dict[str, Any]] None timestamp: str Field(default_factorylambda: datetime.now().isoformat())Pydantic 模型的好处是自带数据校验可以避免脏数据在网络传输中产生隐性错误。3.4 工具注册让 Agent 会“动手”工具是 Agent 与外部世界交互的桥梁。我设计一个简单的Tool数据类它包含工具名称、描述和可调用函数。# 文件路径app/core/tool.py from typing import Any, Callable, Dict, Optional class Tool: def __init__( self, name: str, description: str, func: Callable[..., Any], parameters: Optional[Dict[str, Any]] None, ): self.name name self.description description self.func func self.parameters parameters or {} async def execute(self, **kwargs) - Any: # 统一的工具执行入口 result self.func(**kwargs) return result def to_openai_tool(self) - Dict[str, Any]: 将工具转换为 OpenAI Function Calling 格式。 这样 Agent 可以将工具列表传给 LLM由模型决定调用哪个工具。 return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters, }, }这个设计的关键点是to_openai_tool方法。它把内部工具转换为 LLM 能识别的 JSON Schema 格式这样 LLM 在推理时就能返回结构化的工具调用指令而不是随意写一段文字。3.5 记忆模块Agent 的“数据库”Agent 的记忆分为两层短期记忆存在一个字典里结构是session_id - message_list。长期记忆持久化到 SQLite结构是session_id content。这个原型重点演示短期记忆的接口设计因为长期记忆往往依赖向量数据库会增加不必要的复杂度。# 文件路径app/core/memory.py import json import sqlite3 from typing import Any, Dict, List, Optional class Memory: def __init__(self, db_path: str agent_memory.db): self.db_path db_path self._init_db() # 短期记忆session_id - List[message] self._short_term: Dict[str, List[Dict[str, str]]] {} def _init_db(self): conn sqlite3.connect(self.db_path) conn.execute( CREATE TABLE IF NOT EXISTS long_term_memory ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def add_short_term(self, session_id: str, role: str, content: str) - None: if session_id not in self._short_term: self._short_term[session_id] [] self._short_term[session_id].append( {role: role, content: content} ) def get_short_term(self, session_id: str, limit: int 10) - List[Dict[str, str]]: messages self._short_term.get(session_id, []) return messages[-limit:] def add_long_term(self, session_id: str, role: str, content: str) - None: conn sqlite3.connect(self.db_path) conn.execute( INSERT INTO long_term_memory (session_id, role, content) VALUES (?, ?, ?), (session_id, role, content), ) conn.commit() conn.close() def get_long_term(self, session_id: str, limit: int 20) - List[Dict[str, str]]: conn sqlite3.connect(self.db_path) cursor conn.execute( SELECT role, content FROM long_term_memory WHERE session_id ? ORDER BY id DESC LIMIT ?, (session_id, limit), ) rows cursor.fetchall() conn.close() return [{role: row[0], content: row[1]} for row in rows]记忆模块在真实生产环境中会更复杂例如需要做窗口裁剪、向量化召回、记忆合并去重等。原型阶段我们先把“短期记忆 长期持久化”跑通后续可以平滑替换为更高级的向量记忆。4. 完整实战手写一个可运行的 Agent OS 原型理论部分就到这里接下来是整篇文章最核心的部分我们来实现一个可运行的 Agent OS 原型。这个原型会有两个 Agent、一个调度器、一个 FastAPI 接口能够通过 HTTP 请求来提交任务并获取结果。4.1 创建核心调度器调度器是 Agent OS 的“心脏”。它负责接收任务根据任务类型选择对应的 Agent把任务放入内存队列由 Worker 协程消费队列并执行# 文件路径app/core/scheduler.py import asyncio import logging from typing import Any, Dict, Optional from .agent import BaseAgent from .message import Message, MessageType logger logging.getLogger(__name__) class Scheduler: def __init__(self, registry): self.registry registry self._queue: asyncio.Queue asyncio.Queue() self._workers [] self._running False def start(self, worker_count: int 2): 启动调度器创建指定数量的 Worker 协程。 self._running True for i in range(worker_count): worker asyncio.create_task(self._worker_loop(i 1)) self._workers.append(worker) logger.info(fScheduler started with {worker_count} workers) async def stop(self): 停止调度器等待所有 Worker 完成当前任务。 self._running False for worker in self._workers: worker.cancel() await asyncio.gather(*self._workers, return_exceptionsTrue) logger.info(Scheduler stopped) async def submit(self, agent_name: str, task: str, metadata: Optional[Dict[str, Any]] None) - str: 提交任务到队列返回消息ID。 参数: agent_name: 目标 Agent 名称 task: 任务描述 metadata: 附加元数据如 session_id 返回: 消息ID可通过 get_result 查询结果 msg Message( typeMessageType.TASK, senderscheduler, receiveragent_name, contenttask, metadatametadata or {}, ) await self._queue.put(msg) return msg.msg_id async def _worker_loop(self, worker_id: int): Worker 协程从队列中取出任务找到对应 Agent 并执行。 while self._running: try: msg: Message await self._queue.get() logger.info(f[Worker {worker_id}] received task for {msg.receiver}) agent: Optional[BaseAgent] self.registry.get(msg.receiver) if agent is None: logger.error(fAgent {msg.receiver} 不存在) self._queue.task_done() continue try: await agent.before_run() result await agent.run(msg.content, contextmsg.metadata) # 这里可以扩展把结果持久化到结果表 logger.info( f[Worker {worker_id}] task {msg.msg_id} completed: {result[:50]}... ) except Exception as e: logger.exception(fAgent {msg.receiver} 执行失败: {e}) finally: await agent.after_run() self._queue.task_done() except asyncio.CancelledError: break except Exception as e: logger.exception(fWorker {worker_id} unexpected error: {e}) self._queue.task_done()调度器设计里值得注意的点是submit方法先把任务转成Message对象再入队实现了解耦。任务提交方不需要关心 Agent 在哪里执行、何时执行只需要拿到一个msg_id后续可以通过它轮询结果。4.2 实现第一个 Agent文案写作 Agent现在我们写两个具体的 Agent。先来看文案写作 Agent它的职责是根据主题生成一段营销文案。这个 Agent 需要调用 LLM为了简化实现我直接把 LLM 调用封装在 Agent 内部。在实际项目中你可能会把 LLM 客户端抽成公共依赖而不是在每个 Agent 里重复创建。# 文件路径app/agents/writer_agent.py import asyncio from typing import Any, Dict, Optional from openai import OpenAI from app.core.agent import BaseAgent client OpenAI() class WriterAgent(BaseAgent): def __init__(self): super().__init__( namewriter_agent, description擅长生成营销文案、广告语和产品介绍, system_prompt( 你是一名资深的文案策划专家。 你会根据用户提供的主题生成结构清晰、有吸引力的营销文案。 返回内容使用 Markdown 格式。 ), ) self._model gpt-4o-mini async def run(self, task: str, context: Optional[Dict[str, Any]] None) - str: 生成文案。 # 模拟一个异步等待体现 Agent 执行的时间开销 await asyncio.sleep(0.5) messages [ {role: system, content: self.system_prompt}, {role: user, content: task}, ] # 如果记忆模块存在则先从长期记忆读取历史 if self.memory: history self.memory.get_long_term(context.get(session_id, default)) if history: messages messages[:1] history messages[1:] response client.chat.completions.create( modelself._model, messagesmessages, ) result response.choices[0].message.content # 写入长期记忆 if self.memory: self.memory.add_long_term( context.get(session_id, default), assistant, result ) return result这里有一个工程细节AI Agent 的执行往往不是一次模型调用就能完成的而是“模型推理 → 工具调用 → 结果回填 → 再次推理”的循环。WriterAgent 比较简单一轮调用就能返回结果所以我们没有在代码里做工具调用循环。后面在 Agent 中加入工具能力时需要用while循环处理多个连续的工具调用。4.3 实现第二个 Agent数据分析 Agent第二个 Agent 演示工具调用能力。它有一个calculate工具可以执行简单的数学运算。这个例子虽然简化但展示了“Agent 工具”的组合方式。# 文件路径app/agents/analysis_agent.py import json from typing import Any, Dict, Optional from openai import OpenAI from app.core.agent import BaseAgent from app.core.tool import Tool client OpenAI() def calculate(expression: str) - str: 计算一个数学表达式的值。 注意这里只允许执行简单的加减乘除运算不允许执行任意 Python 代码。 allowed_chars set(0123456789-*/(). ) if not set(expression).issubset(allowed_chars): return 错误表达式包含非法字符 try: # 使用 eval 有安全风险示例中仅演示生产环境请使用更安全的计算方案 result eval(expression) return str(result) except Exception as e: return f计算错误: {str(e)} class AnalysisAgent(BaseAgent): def __init__(self): calc_tool Tool( namecalculate, description计算数学表达式的值例如 1 2 * 3, funccalculate, parameters{ type: object, properties: { expression: { type: string, description: 要计算的数学表达式, } }, required: [expression], }, ) super().__init__( nameanalysis_agent, description擅长数据计算和简单分析, system_prompt( 你是一个数据分析助手。 当用户需要计算时你必须使用 calculate 工具。 计算完成后用中文解释计算结果。 ), tools[calc_tool], ) self._model gpt-4o-mini async def run(self, task: str, context: Optional[Dict[str, Any]] None) - str: 执行任务。这里演示一个简化的工具调用循环。 await asyncio.sleep(0.5) messages [ {role: system, content: self.system_prompt}, {role: user, content: task}, ] # 将工具转换为 OpenAI 格式 openai_tools [tool.to_openai_tool() for tool in self.tools] # 最多执行 5 轮工具调用避免无限循环 for _ in range(5): response client.chat.completions.create( modelself._model, messagesmessages, toolsopenai_tools, tool_choiceauto, ) choice response.choices[0] message choice.message if not message.tool_calls: # 模型不再调用工具直接返回结果 return message.content # 处理工具调用 messages.append(message) for tool_call in message.tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) # 找到对应的 Tool 并执行 tool next(t for t in self.tools if t.name tool_name) tool_result await tool.execute(**tool_args) messages.append( { role: tool, tool_call_id: tool_call.id, content: tool_result, } ) return 任务执行超时未在限定轮次内完成analysis_agent里的工具调用循环是 Agent 系统里非常核心的代码模式。它让模型可以“用自然语言思考用工具行动”并且能够在多个工具之间来回切换逐步逼近最终答案。需要注意的是eval的安全性。这里我做了字符白名单校验仅允许数字、运算符和括号避免执行任意代码。生产环境务必要把eval替换为ast.literal_eval或专门的表达式解析库例如asteval。4.4 组装 FastAPI 入口有了 Agent、注册表、调度器接下来把它们组装成一个 Web 服务。FastAPI 在这里的角色是 HTTP 入口层把外部请求转化为内部任务。# 文件路径app/main.py import asyncio from contextlib import asynccontextmanager from typing import Optional from fastapi import FastAPI, HTTPException from pydantic import BaseModel from app.core.agent import BaseAgent from app.core.memory import Memory from app.core.registry import AgentRegistry from app.core.scheduler import Scheduler from app.agents.writer_agent import WriterAgent from app.agents.analysis_agent import AnalysisAgent # 全局变量 registry AgentRegistry() memory Memory(db_pathagent_memory.db) scheduler None asynccontextmanager async def lifespan(app: FastAPI): global scheduler # 创建 Agent 并注册 writer WriterAgent() writer.memory memory analysis AnalysisAgent() analysis.memory memory registry.register(writer) registry.register(analysis) # 创建并启动调度器 scheduler Scheduler(registry) scheduler.start(worker_count2) print(Agent OS 启动完成当前可用 Agent) for agent in registry.list_agents(): print(f - {agent[name]}: {agent[description]}) yield # 关闭调度器 await scheduler.stop() app FastAPI(titleAgent OS Demo, lifespanlifespan) class TaskRequest(BaseModel): agent_name: str task: str session_id: Optional[str] default class TaskResponse(BaseModel): msg_id: str status: str submitted app.get(/agents) async def list_agents(): return registry.list_agents() app.post(/task, response_modelTaskResponse) async def submit_task(req: TaskRequest): 提交一个任务给指定的 Agent。这是一个异步接口 调用方拿到 msg_id 后可以后续查询结果。 注意这个原型没有实现结果查询接口实际项目中需要 增加一个 result_store 来保存执行结果。 agent registry.get(req.agent_name) if agent is None: raise HTTPException(status_code404, detailfAgent {req.agent_name} 不存在) msg_id await scheduler.submit( agent_namereq.agent_name, taskreq.task, metadata{session_id: req.session_id}, ) return TaskResponse(msg_idmsg_id)注意lifespan里的初始化逻辑先创建 Agent注入记忆对象注册再启动调度器。这种顺序很重要因为调度器的 Worker 会从队列里取任务而 Agent 必须先注册完成才能被 Worker 找到。4.5 运行与验证现在来启动服务。在项目根目录执行uvicorn app.main:app --reload --host 0.0.0.0 --port 8000启动后你应该看到类似下面的日志INFO: Uvicorn running on http://0.0.0.0:8000 INFO: Started server process [12345] INFO: Waiting for application startup. Agent OS 启动完成当前可用 Agent - writer_agent: 擅长生成营销文案、广告语和产品介绍 - analysis_agent: 擅长数据计算和简单分析 INFO: Application startup complete.然后可以用curl来测试。查看所有 Agentcurl http://localhost:8000/agents预期输出[ { name: writer_agent, description: 擅长生成营销文案、广告语和产品介绍 }, { name: analysis_agent, description: 擅长数据计算和简单分析 } ]提交一个文案任务curl -X POST http://localhost:8000/task \ -H Content-Type: application/json \ -d {agent_name: writer_agent, task: 为一家咖啡店写一句宣传语, session_id: test-session-001}预期输出{ msg_id: msg_1718000000.123, status: submitted }再提交一个计算任务curl -X POST http://localhost:8000/task \ -H Content-Type: application/json \ -d {agent_name: analysis_agent, task: 计算 (12 8) * 3 的结果, session_id: test-session-001}你会在终端日志里看到 Worker 消费任务的记录。由于原型没有实现结果存储接口Agent 的实际输出只能从日志中查看。如果你希望让结果也能通过 HTTP 查询一个简单的升级方案是在Scheduler里增加一个result_store: Dict[str, str]在_worker_loop中把agent.run()的返回值存入字典再增加一个/task/{msg_id}接口查询结果。这个升级非常适合作为练习自己实现。5. 常见问题与排查思路在实际运行这个 Agent OS 原型时你可能会遇到下面这些问题。我把它们的现象、触发原因和排查方法整理成表格方便快速定位。问题现象常见原因解决思路uvicorn启动失败端口被占用换一个端口例如--port 8001或先找到占用进程后结束ImportError: No module named openai依赖没有安装完整执行pip install -r requirements.txt重点确认 openai 已安装Agent xxx 已存在重复注册同名 Agent检查注册代码是否在 lifespan 中被调用了多次确认 Agent 名称唯一提交任务后无任何日志输出调度器没有启动检查lifespan中是否调用了scheduler.start()确认 Worker 数量大于 0Agent 返回结果与预期不符系统提示词不够明确或模型版本有问题优化system_prompt补充输出格式约束尝试更换模型工具调用循环无法结束模型反复调用同一个工具或工具返回值不满足模型预期设置最大循环次数本示例为 5 次并对工具返回值做结构化处理SQLite 数据库文件报错多进程并发写入冲突原型阶段使用单进程运行生产环境替换为 PostgreSQL 或 MySQL中文乱码终端编码不是 UTF-8Windows 终端执行chcp 65001macOS/Linux 一般无此问题这里重点说一下“工具调用循环无法结束”这个问题。它很常见根因通常是工具返回的内容太“模糊”。比如calculate函数返回的是计算错误: invalid syntax模型可能会误以为用户输入格式有问题然后反复尝试传类似参数。解决方案是让工具返回值尽量包含足够的信息例如在错误信息中附带“请检查表达式是否包含非数字字符”的提示帮助模型自我纠正。6. 最佳实践与工程建议有了一个可运行的原型后我们再从工程化的角度讨论如何把这个“玩具系统”变成一个真正可以上生产的 Agent OS。下面五个方向是我在实战中认为优先级最高的。6.1 Agent 名称与职责边界要清晰Agent 的命名应体现“职责”而不是“功能”。比如writer_agent、analysis_agent这种名称比agent1、agent2容易理解和维护。更重要的是一个 Agent 只负责一个领域的任务不要让一个 Agent 既写文案又做财务核算又发邮件。职责单一的好处是系统提示词容易维护工具权限容易收敛出问题时排查范围也小。6.2 配置与提示词分离我在原型里把system_prompt硬编码在 Agent 类里这在项目初期没问题但一旦 Agent 数量增多你需要把提示词放到配置文件或外部存储中实现提示词的动态更新和灰度发布。建议使用 YAML 或 JSON 文件集中管理提示词模板这样产品人员也能参与维护。6.3 安全边界必须从第一天就考虑Agent 调用工具时必须遵循最小权限原则。不要给 Agent 一个可以执行任意 Python 代码的工具。如果 Agent 需要访问数据库创建专用账号只授予所需表的 SELECT、INSERT 权限。如果 Agent 需要调用第三方 API使用单独的 API Key并设置配额限制。对 Agent 的工具调用日志做全量记录包括工具参数和执行结果便于安全审计。在原型里我对calculate工具做了字符白名单校验这就是一种安全收敛。生产环境上你可以使用专门的沙箱机制如 Docker 隔离、子进程执行来执行不可信代码。6.4 可观测性是 Agent OS 的生命线Agent 系统比普通 Web 系统更难排查问题因为它的执行路径往往跨越多次模型推理和多次工具调用。强烈建议从第一天就建立日志规范。每个任务至少要记录以下信息任务 ID 与父任务 ID用于追踪任务树Agent 名称输入消息和输出消息每次 LLM 调用的 token 开销工具调用的入参和返回值异常堆栈在原型里我们用了 Python 自带的logging这还不够。生产环境建议接入 OpenTelemetry 或 Langfuse 这类可观测平台将 Agent 的执行链路可视化。6.5 结果与状态管理要尽早设计原型里没有实现结果查询接口这是一个明显的短板。真实项目中任务提交后通常需要支持以下操作查询任务状态running / completed / failed查询任务执行结果取消一个任务重试一个失败的任务这些能力需要你设计一个TaskStore把任务的状态、输入、输出持久化到数据库。这也是 Agent OS 从“原型”走向“产品”的关键一步。7. 总结与下一步学习路线这篇文章从 Agent OS 的概念讲起解释了它为什么不是传统意义上的操作系统而是智能体运行时所需的一组基础设施抽象。然后我们通过一个完整的 Python 原型实现了 Agent 注册、任务调度、消息传递、记忆存储、工具调用等核心能力。通过这个实战项目你应该掌握了以下关键技能理解 Agent 的生命周期状态与状态管理方式。掌握 Agent 注册表模式能够在系统中统一管理多个 Agent。理解任务调度器的基本原理知道如何用asyncio.Queue实现简单的异步任务分发。掌握 Agent 工具调用的开发模式了解模型如何依据工具 Schema 生成调用指令。理解记忆模块的分层设计短期记忆与长期持久化各自的定位。下一步你可以从以下几个方向继续深入实现TaskStore支持任务结果查询与失败重试。将内存队列替换为 Redis Stream实现多实例部署。引入向量数据库为 Agent 添加语义检索长期记忆。研究多 Agent 编排模式例如“规划 Agent 拆解任务 → 执行 Agent 实际操作 → 汇总 Agent 输出”。学习 LangGraph、AutoGen、CrewAI 等成熟的 Agent 编排框架对照本文原型的模块理解成熟框架的设计取舍。如果这篇文章对你有帮助建议收藏备用后续你开发 Agent 应用时可以随时回来对照架构设计和代码思路。动手把原型跑起来才是掌握 Agent OS 核心思想的最好方式。 从零手写轻量级Agent OS：AI智能体运行时框架实战
你好我是资深技术博主。今天我们不聊理论空谈直接围绕“Agent OS”这个热门概念拆解它的核心本质并带你手写一个可运行的轻量级 Agent 运行时框架。先说结论Agent OS 不是一个具体的开源软件也不是某个大厂发布的新操作系统。它是一套面向 AI Agent智能体的设计范式与运行时架构。它的目标是解决一个真实痛点当你的项目里只有一个 LLM 调用时不需要复杂的调度但当你有几十个、上百个 Agent 协同工作需要管理记忆、工具、任务队列、上下文窗口和异常恢复时普通脚本已经无法胜任。这篇文章会从零开始带你理解 Agent OS 的核心概念然后搭建一个具备“Agent 注册、任务调度、工具调用、消息通信、记忆存储”五个核心能力的简化版 Agent OS。整个过程我会尽量用完整的代码和配置示例让你照着敲就能跑起来。如果你是一个刚接触 LLM 应用的开发者本文能帮你建立 Agent 系统设计的整体认知如果你已经在做 Agent 应用本文提供的架构拆分和工程化思路可以直接迁移到你的项目中。1. Agent OS 到底是什么先建立整体认知1.1 从“单 Agent 脚本”到“Agent OS”的演进我们先用一个简单的对比来理解。最原始的 LLM 应用很简单就是一个函数调用from openai import OpenAI client OpenAI() def ask_llm(prompt: str) - str: response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], ) return response.choices[0].message.content print(ask_llm(你好介绍一下你自己))这个脚本一旦跑起来就是一个线性流程用户输入 → 模型推理 → 输出结果。它没有状态没有记忆没有工具没有多 Agent 协作。到第二阶段开发者会在脚本里加入工具调用Function Calling。模型在推理时可能返回一个“调用某个函数”的指令脚本会执行这个函数把结果再传回给模型。tools [ { type: function, function: { name: get_weather, description: 获取指定城市的天气, parameters: { type: object, properties: { city: {type: string} }, required: [city] } } } ]这个阶段还是单 Agent 的脚本化增强。到了第三个阶段当你的业务足够复杂比如需要“一个 Agent 负责数据分析一个 Agent 负责生成报告一个 Agent 负责发送邮件”并且这些 Agent 之间还需要互相传递上下文、共享记忆、按优先级排队执行时你就需要一套能够统一管理它们的运行时环境。这套运行时的角色类似于传统服务器里操作系统的作用传统操作系统Agent OS 对应能力进程管理Agent 生命周期管理内存管理上下文窗口与记忆管理进程间通信IPCAgent 之间的消息传递文件系统长期记忆与知识存储设备驱动工具Tools插件机制调度器任务队列与优先级调度系统日志推理过程监控与追踪这就是 Agent OS 的概念来源。它不是一个具体的软件产品而是一套让 Agent 系统在生产环境中稳定运行所需的基础设施抽象。1.2 Agent OS 解决的四类核心问题通过上面的类比你应该已经理解了 Agent OS 的定位。接下来我们把它要解决的具体问题归纳成四类这四类问题也是后面实战部分的设计依据。第一类会话生命周期管理单个 Agent 请求是无状态的但业务是有状态的。你需要清楚一个任务从进入队列到最终完成经历了哪些阶段一个 Agent 在运行中被中断后如何恢复一个长时任务是否需要暂停和续跑。Agent OS 需要定义清晰的状态机并统一管理这些状态转换。第二类记忆与上下文管理LLM 的上下文窗口是有限的。你不能把所有历史消息都塞进每一次请求。Agent OS 需要设计分级记忆机制比如短期记忆当前会话上下文、长期记忆持久化到向量数据库或数据库、工作记忆当前任务相关的临时数据。第三类工具注册与动态调用Agent 要完成真实世界的任务必须调用外部工具。工具可能是内部函数、REST API、数据库操作等。Agent OS 需要提供统一的工具注册表让 Agent 能够发现工具、了解工具的入参格式并在运行时安全地调用工具。第四类多 Agent 协作与任务分发复杂任务往往需要拆分成多个子任务分配给不同的 Agent 并行或串行执行。Agent OS 需要支持任务拆分、结果汇总、消息传递、冲突协调等能力。1.3 Agent OS 一定需要一个“操作系统”吗这里有一个常见误区并不是所有 Agent 应用都需要一套完整的 Agent OS。对于只有几个 Agent、任务流程固定、不需要长期记忆的小项目用简单的编排框架就足够了。Agent OS 的价值在系统规模变大后才体现出来。这篇文章的实战案例会设计一个“面向中小型项目”的轻量级 Agent OS 框架。虽然叫 OS但它的实现并不复杂核心就是把上面提到的五类能力抽象成五个模块用 FastAPI 加 SQLite 就能支撑一个可用的原型。2. 环境准备与项目初始化在正式写代码之前先来准备环境。由于 Agent OS 本身是一个 Python 后端服务我们需要配置好 Python 环境和相关依赖。2.1 版本说明为了保持通用性这里不依赖某个特定的 LLM 厂商 SDK而是通过统一接口来调用大模型。你可以根据实际情况替换为 OpenAI、通义千问、文心一言或本地部署的 Ollama 模型。示例环境如下操作系统Windows 10 / macOS / Linux 均可Python3.10 或以上版本数据库SQLitePython 内置Web 框架FastAPI消息队列使用 Python 内置的asyncio.Queue做内存队列生产环境可替换为 Redis Stream 或 RabbitMQ2.2 创建项目结构先创建项目目录我建议使用下面的结构清晰且容易扩展agent-os-demo/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── core/ │ │ ├── __init__.py │ │ ├── agent.py # Agent 抽象类 │ │ ├── registry.py # Agent 注册表 │ │ ├── message.py # 消息模型 │ │ ├── scheduler.py # 任务调度器 │ │ ├── memory.py # 记忆存储 │ │ └── tool.py # 工具注册 │ └── agents/ │ ├── __init__.py │ ├── writer_agent.py # 示例 Agent文案写作 │ └── analysis_agent.py # 示例 Agent数据分析 ├── requirements.txt └── README.md2.3 安装依赖创建requirements.txt内容如下fastapi0.111.0 uvicorn0.30.1 openai1.30.0 pydantic2.7.1然后在终端执行pip install -r requirements.txt这里需要注意openai库并不是 Agent OS 的必需依赖但示例代码需要一个稳定的 LLM 调用方式来演示。你可以把它替换成任何你熟悉的大模型 SDK接口不变。3. 核心模块设计与原理拆解在写代码之前我们先花一些时间理解每个核心模块的设计思路。这样后面贴完整代码时你能清楚每一段代码在系统里扮演什么角色。3.1 Agent 抽象Agent 到底是什么在 Agent OS 框架里Agent 不是一个大模型而是“大模型 指令模板 工具权限 上下文策略”的组合体。我定义一个BaseAgent抽象类它具备以下属性nameAgent 的唯一名称用于注册和路由。description描述这个 Agent 擅长什么任务供调度器做任务分配。system_prompt系统提示词定义 Agent 的角色和行为规范。tools该 Agent 可以调用的工具列表。memory该 Agent 使用的记忆对象。每个 Agent 的核心方法是run(task: str, context: dict) - str调度器通过这个方法把任务交给 Agent 执行。# 文件路径app/core/agent.py from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional class BaseAgent(ABC): def __init__( self, name: str, description: str, system_prompt: str, tools: Optional[List[Any]] None, memory: Optional[Any] None, ): self.name name self.description description self.system_prompt system_prompt self.tools tools or [] self.memory memory self.status idle # idle - running - finished self.max_retry 3 abstractmethod async def run(self, task: str, context: Optional[Dict[str, Any]] None) - str: 执行一个具体任务。 参数: task: 任务描述 context: 额外的上下文信息比如用户ID、会话ID 返回: Agent 执行结果字符串 pass async def before_run(self): self.status running async def after_run(self): self.status finished def __repr__(self): return fAgent name{self.name} status{self.status}这里把run定义成异步方法是因为真实的 Agent 任务往往涉及网络调用、工具执行异步能更好地利用系统资源。3.2 注册表Agent 的“电话簿”注册表的作用是让系统知道有哪些 Agent 可用。它提供三个关键能力register注册一个 Agent。get根据名称获取 Agent。list_agents列出所有 Agent 的基本信息。# 文件路径app/core/registry.py from typing import Dict, Optional from .agent import BaseAgent class AgentRegistry: def __init__(self): self._agents: Dict[str, BaseAgent] {} def register(self, agent: BaseAgent) - None: if agent.name in self._agents: raise ValueError(fAgent {agent.name} 已存在请更换名称) self._agents[agent.name] agent def get(self, name: str) - Optional[BaseAgent]: return self._agents.get(name) def list_agents(self) - list: return [ {name: name, description: agent.description} for name, agent in self._agents.items() ]注册表设计模式的要点是“名称唯一”。在大型系统中Agent 名称唯一是路由和调度的基础避免多个同名 Agent 导致任务投递不明确。3.3 消息模型Agent 之间怎么说话在多 Agent 协作中消息模型决定了系统能否扩展。我把消息分成两种AgentMessageAgent 之间的通信消息TaskMessage调度器下发给 Agent 的任务消息# 文件路径app/core/message.py from enum import Enum from typing import Any, Dict, Optional from pydantic import BaseModel, Field from datetime import datetime class MessageType(str, Enum): TASK task RESULT result ERROR error HEARTBEAT heartbeat class Message(BaseModel): msg_id: str Field(default_factorylambda: fmsg_{datetime.now().timestamp()}) type: MessageType sender: str receiver: str content: str metadata: Optional[Dict[str, Any]] None timestamp: str Field(default_factorylambda: datetime.now().isoformat())Pydantic 模型的好处是自带数据校验可以避免脏数据在网络传输中产生隐性错误。3.4 工具注册让 Agent 会“动手”工具是 Agent 与外部世界交互的桥梁。我设计一个简单的Tool数据类它包含工具名称、描述和可调用函数。# 文件路径app/core/tool.py from typing import Any, Callable, Dict, Optional class Tool: def __init__( self, name: str, description: str, func: Callable[..., Any], parameters: Optional[Dict[str, Any]] None, ): self.name name self.description description self.func func self.parameters parameters or {} async def execute(self, **kwargs) - Any: # 统一的工具执行入口 result self.func(**kwargs) return result def to_openai_tool(self) - Dict[str, Any]: 将工具转换为 OpenAI Function Calling 格式。 这样 Agent 可以将工具列表传给 LLM由模型决定调用哪个工具。 return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters, }, }这个设计的关键点是to_openai_tool方法。它把内部工具转换为 LLM 能识别的 JSON Schema 格式这样 LLM 在推理时就能返回结构化的工具调用指令而不是随意写一段文字。3.5 记忆模块Agent 的“数据库”Agent 的记忆分为两层短期记忆存在一个字典里结构是session_id - message_list。长期记忆持久化到 SQLite结构是session_id content。这个原型重点演示短期记忆的接口设计因为长期记忆往往依赖向量数据库会增加不必要的复杂度。# 文件路径app/core/memory.py import json import sqlite3 from typing import Any, Dict, List, Optional class Memory: def __init__(self, db_path: str agent_memory.db): self.db_path db_path self._init_db() # 短期记忆session_id - List[message] self._short_term: Dict[str, List[Dict[str, str]]] {} def _init_db(self): conn sqlite3.connect(self.db_path) conn.execute( CREATE TABLE IF NOT EXISTS long_term_memory ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def add_short_term(self, session_id: str, role: str, content: str) - None: if session_id not in self._short_term: self._short_term[session_id] [] self._short_term[session_id].append( {role: role, content: content} ) def get_short_term(self, session_id: str, limit: int 10) - List[Dict[str, str]]: messages self._short_term.get(session_id, []) return messages[-limit:] def add_long_term(self, session_id: str, role: str, content: str) - None: conn sqlite3.connect(self.db_path) conn.execute( INSERT INTO long_term_memory (session_id, role, content) VALUES (?, ?, ?), (session_id, role, content), ) conn.commit() conn.close() def get_long_term(self, session_id: str, limit: int 20) - List[Dict[str, str]]: conn sqlite3.connect(self.db_path) cursor conn.execute( SELECT role, content FROM long_term_memory WHERE session_id ? ORDER BY id DESC LIMIT ?, (session_id, limit), ) rows cursor.fetchall() conn.close() return [{role: row[0], content: row[1]} for row in rows]记忆模块在真实生产环境中会更复杂例如需要做窗口裁剪、向量化召回、记忆合并去重等。原型阶段我们先把“短期记忆 长期持久化”跑通后续可以平滑替换为更高级的向量记忆。4. 完整实战手写一个可运行的 Agent OS 原型理论部分就到这里接下来是整篇文章最核心的部分我们来实现一个可运行的 Agent OS 原型。这个原型会有两个 Agent、一个调度器、一个 FastAPI 接口能够通过 HTTP 请求来提交任务并获取结果。4.1 创建核心调度器调度器是 Agent OS 的“心脏”。它负责接收任务根据任务类型选择对应的 Agent把任务放入内存队列由 Worker 协程消费队列并执行# 文件路径app/core/scheduler.py import asyncio import logging from typing import Any, Dict, Optional from .agent import BaseAgent from .message import Message, MessageType logger logging.getLogger(__name__) class Scheduler: def __init__(self, registry): self.registry registry self._queue: asyncio.Queue asyncio.Queue() self._workers [] self._running False def start(self, worker_count: int 2): 启动调度器创建指定数量的 Worker 协程。 self._running True for i in range(worker_count): worker asyncio.create_task(self._worker_loop(i 1)) self._workers.append(worker) logger.info(fScheduler started with {worker_count} workers) async def stop(self): 停止调度器等待所有 Worker 完成当前任务。 self._running False for worker in self._workers: worker.cancel() await asyncio.gather(*self._workers, return_exceptionsTrue) logger.info(Scheduler stopped) async def submit(self, agent_name: str, task: str, metadata: Optional[Dict[str, Any]] None) - str: 提交任务到队列返回消息ID。 参数: agent_name: 目标 Agent 名称 task: 任务描述 metadata: 附加元数据如 session_id 返回: 消息ID可通过 get_result 查询结果 msg Message( typeMessageType.TASK, senderscheduler, receiveragent_name, contenttask, metadatametadata or {}, ) await self._queue.put(msg) return msg.msg_id async def _worker_loop(self, worker_id: int): Worker 协程从队列中取出任务找到对应 Agent 并执行。 while self._running: try: msg: Message await self._queue.get() logger.info(f[Worker {worker_id}] received task for {msg.receiver}) agent: Optional[BaseAgent] self.registry.get(msg.receiver) if agent is None: logger.error(fAgent {msg.receiver} 不存在) self._queue.task_done() continue try: await agent.before_run() result await agent.run(msg.content, contextmsg.metadata) # 这里可以扩展把结果持久化到结果表 logger.info( f[Worker {worker_id}] task {msg.msg_id} completed: {result[:50]}... ) except Exception as e: logger.exception(fAgent {msg.receiver} 执行失败: {e}) finally: await agent.after_run() self._queue.task_done() except asyncio.CancelledError: break except Exception as e: logger.exception(fWorker {worker_id} unexpected error: {e}) self._queue.task_done()调度器设计里值得注意的点是submit方法先把任务转成Message对象再入队实现了解耦。任务提交方不需要关心 Agent 在哪里执行、何时执行只需要拿到一个msg_id后续可以通过它轮询结果。4.2 实现第一个 Agent文案写作 Agent现在我们写两个具体的 Agent。先来看文案写作 Agent它的职责是根据主题生成一段营销文案。这个 Agent 需要调用 LLM为了简化实现我直接把 LLM 调用封装在 Agent 内部。在实际项目中你可能会把 LLM 客户端抽成公共依赖而不是在每个 Agent 里重复创建。# 文件路径app/agents/writer_agent.py import asyncio from typing import Any, Dict, Optional from openai import OpenAI from app.core.agent import BaseAgent client OpenAI() class WriterAgent(BaseAgent): def __init__(self): super().__init__( namewriter_agent, description擅长生成营销文案、广告语和产品介绍, system_prompt( 你是一名资深的文案策划专家。 你会根据用户提供的主题生成结构清晰、有吸引力的营销文案。 返回内容使用 Markdown 格式。 ), ) self._model gpt-4o-mini async def run(self, task: str, context: Optional[Dict[str, Any]] None) - str: 生成文案。 # 模拟一个异步等待体现 Agent 执行的时间开销 await asyncio.sleep(0.5) messages [ {role: system, content: self.system_prompt}, {role: user, content: task}, ] # 如果记忆模块存在则先从长期记忆读取历史 if self.memory: history self.memory.get_long_term(context.get(session_id, default)) if history: messages messages[:1] history messages[1:] response client.chat.completions.create( modelself._model, messagesmessages, ) result response.choices[0].message.content # 写入长期记忆 if self.memory: self.memory.add_long_term( context.get(session_id, default), assistant, result ) return result这里有一个工程细节AI Agent 的执行往往不是一次模型调用就能完成的而是“模型推理 → 工具调用 → 结果回填 → 再次推理”的循环。WriterAgent 比较简单一轮调用就能返回结果所以我们没有在代码里做工具调用循环。后面在 Agent 中加入工具能力时需要用while循环处理多个连续的工具调用。4.3 实现第二个 Agent数据分析 Agent第二个 Agent 演示工具调用能力。它有一个calculate工具可以执行简单的数学运算。这个例子虽然简化但展示了“Agent 工具”的组合方式。# 文件路径app/agents/analysis_agent.py import json from typing import Any, Dict, Optional from openai import OpenAI from app.core.agent import BaseAgent from app.core.tool import Tool client OpenAI() def calculate(expression: str) - str: 计算一个数学表达式的值。 注意这里只允许执行简单的加减乘除运算不允许执行任意 Python 代码。 allowed_chars set(0123456789-*/(). ) if not set(expression).issubset(allowed_chars): return 错误表达式包含非法字符 try: # 使用 eval 有安全风险示例中仅演示生产环境请使用更安全的计算方案 result eval(expression) return str(result) except Exception as e: return f计算错误: {str(e)} class AnalysisAgent(BaseAgent): def __init__(self): calc_tool Tool( namecalculate, description计算数学表达式的值例如 1 2 * 3, funccalculate, parameters{ type: object, properties: { expression: { type: string, description: 要计算的数学表达式, } }, required: [expression], }, ) super().__init__( nameanalysis_agent, description擅长数据计算和简单分析, system_prompt( 你是一个数据分析助手。 当用户需要计算时你必须使用 calculate 工具。 计算完成后用中文解释计算结果。 ), tools[calc_tool], ) self._model gpt-4o-mini async def run(self, task: str, context: Optional[Dict[str, Any]] None) - str: 执行任务。这里演示一个简化的工具调用循环。 await asyncio.sleep(0.5) messages [ {role: system, content: self.system_prompt}, {role: user, content: task}, ] # 将工具转换为 OpenAI 格式 openai_tools [tool.to_openai_tool() for tool in self.tools] # 最多执行 5 轮工具调用避免无限循环 for _ in range(5): response client.chat.completions.create( modelself._model, messagesmessages, toolsopenai_tools, tool_choiceauto, ) choice response.choices[0] message choice.message if not message.tool_calls: # 模型不再调用工具直接返回结果 return message.content # 处理工具调用 messages.append(message) for tool_call in message.tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) # 找到对应的 Tool 并执行 tool next(t for t in self.tools if t.name tool_name) tool_result await tool.execute(**tool_args) messages.append( { role: tool, tool_call_id: tool_call.id, content: tool_result, } ) return 任务执行超时未在限定轮次内完成analysis_agent里的工具调用循环是 Agent 系统里非常核心的代码模式。它让模型可以“用自然语言思考用工具行动”并且能够在多个工具之间来回切换逐步逼近最终答案。需要注意的是eval的安全性。这里我做了字符白名单校验仅允许数字、运算符和括号避免执行任意代码。生产环境务必要把eval替换为ast.literal_eval或专门的表达式解析库例如asteval。4.4 组装 FastAPI 入口有了 Agent、注册表、调度器接下来把它们组装成一个 Web 服务。FastAPI 在这里的角色是 HTTP 入口层把外部请求转化为内部任务。# 文件路径app/main.py import asyncio from contextlib import asynccontextmanager from typing import Optional from fastapi import FastAPI, HTTPException from pydantic import BaseModel from app.core.agent import BaseAgent from app.core.memory import Memory from app.core.registry import AgentRegistry from app.core.scheduler import Scheduler from app.agents.writer_agent import WriterAgent from app.agents.analysis_agent import AnalysisAgent # 全局变量 registry AgentRegistry() memory Memory(db_pathagent_memory.db) scheduler None asynccontextmanager async def lifespan(app: FastAPI): global scheduler # 创建 Agent 并注册 writer WriterAgent() writer.memory memory analysis AnalysisAgent() analysis.memory memory registry.register(writer) registry.register(analysis) # 创建并启动调度器 scheduler Scheduler(registry) scheduler.start(worker_count2) print(Agent OS 启动完成当前可用 Agent) for agent in registry.list_agents(): print(f - {agent[name]}: {agent[description]}) yield # 关闭调度器 await scheduler.stop() app FastAPI(titleAgent OS Demo, lifespanlifespan) class TaskRequest(BaseModel): agent_name: str task: str session_id: Optional[str] default class TaskResponse(BaseModel): msg_id: str status: str submitted app.get(/agents) async def list_agents(): return registry.list_agents() app.post(/task, response_modelTaskResponse) async def submit_task(req: TaskRequest): 提交一个任务给指定的 Agent。这是一个异步接口 调用方拿到 msg_id 后可以后续查询结果。 注意这个原型没有实现结果查询接口实际项目中需要 增加一个 result_store 来保存执行结果。 agent registry.get(req.agent_name) if agent is None: raise HTTPException(status_code404, detailfAgent {req.agent_name} 不存在) msg_id await scheduler.submit( agent_namereq.agent_name, taskreq.task, metadata{session_id: req.session_id}, ) return TaskResponse(msg_idmsg_id)注意lifespan里的初始化逻辑先创建 Agent注入记忆对象注册再启动调度器。这种顺序很重要因为调度器的 Worker 会从队列里取任务而 Agent 必须先注册完成才能被 Worker 找到。4.5 运行与验证现在来启动服务。在项目根目录执行uvicorn app.main:app --reload --host 0.0.0.0 --port 8000启动后你应该看到类似下面的日志INFO: Uvicorn running on http://0.0.0.0:8000 INFO: Started server process [12345] INFO: Waiting for application startup. Agent OS 启动完成当前可用 Agent - writer_agent: 擅长生成营销文案、广告语和产品介绍 - analysis_agent: 擅长数据计算和简单分析 INFO: Application startup complete.然后可以用curl来测试。查看所有 Agentcurl http://localhost:8000/agents预期输出[ { name: writer_agent, description: 擅长生成营销文案、广告语和产品介绍 }, { name: analysis_agent, description: 擅长数据计算和简单分析 } ]提交一个文案任务curl -X POST http://localhost:8000/task \ -H Content-Type: application/json \ -d {agent_name: writer_agent, task: 为一家咖啡店写一句宣传语, session_id: test-session-001}预期输出{ msg_id: msg_1718000000.123, status: submitted }再提交一个计算任务curl -X POST http://localhost:8000/task \ -H Content-Type: application/json \ -d {agent_name: analysis_agent, task: 计算 (12 8) * 3 的结果, session_id: test-session-001}你会在终端日志里看到 Worker 消费任务的记录。由于原型没有实现结果存储接口Agent 的实际输出只能从日志中查看。如果你希望让结果也能通过 HTTP 查询一个简单的升级方案是在Scheduler里增加一个result_store: Dict[str, str]在_worker_loop中把agent.run()的返回值存入字典再增加一个/task/{msg_id}接口查询结果。这个升级非常适合作为练习自己实现。5. 常见问题与排查思路在实际运行这个 Agent OS 原型时你可能会遇到下面这些问题。我把它们的现象、触发原因和排查方法整理成表格方便快速定位。问题现象常见原因解决思路uvicorn启动失败端口被占用换一个端口例如--port 8001或先找到占用进程后结束ImportError: No module named openai依赖没有安装完整执行pip install -r requirements.txt重点确认 openai 已安装Agent xxx 已存在重复注册同名 Agent检查注册代码是否在 lifespan 中被调用了多次确认 Agent 名称唯一提交任务后无任何日志输出调度器没有启动检查lifespan中是否调用了scheduler.start()确认 Worker 数量大于 0Agent 返回结果与预期不符系统提示词不够明确或模型版本有问题优化system_prompt补充输出格式约束尝试更换模型工具调用循环无法结束模型反复调用同一个工具或工具返回值不满足模型预期设置最大循环次数本示例为 5 次并对工具返回值做结构化处理SQLite 数据库文件报错多进程并发写入冲突原型阶段使用单进程运行生产环境替换为 PostgreSQL 或 MySQL中文乱码终端编码不是 UTF-8Windows 终端执行chcp 65001macOS/Linux 一般无此问题这里重点说一下“工具调用循环无法结束”这个问题。它很常见根因通常是工具返回的内容太“模糊”。比如calculate函数返回的是计算错误: invalid syntax模型可能会误以为用户输入格式有问题然后反复尝试传类似参数。解决方案是让工具返回值尽量包含足够的信息例如在错误信息中附带“请检查表达式是否包含非数字字符”的提示帮助模型自我纠正。6. 最佳实践与工程建议有了一个可运行的原型后我们再从工程化的角度讨论如何把这个“玩具系统”变成一个真正可以上生产的 Agent OS。下面五个方向是我在实战中认为优先级最高的。6.1 Agent 名称与职责边界要清晰Agent 的命名应体现“职责”而不是“功能”。比如writer_agent、analysis_agent这种名称比agent1、agent2容易理解和维护。更重要的是一个 Agent 只负责一个领域的任务不要让一个 Agent 既写文案又做财务核算又发邮件。职责单一的好处是系统提示词容易维护工具权限容易收敛出问题时排查范围也小。6.2 配置与提示词分离我在原型里把system_prompt硬编码在 Agent 类里这在项目初期没问题但一旦 Agent 数量增多你需要把提示词放到配置文件或外部存储中实现提示词的动态更新和灰度发布。建议使用 YAML 或 JSON 文件集中管理提示词模板这样产品人员也能参与维护。6.3 安全边界必须从第一天就考虑Agent 调用工具时必须遵循最小权限原则。不要给 Agent 一个可以执行任意 Python 代码的工具。如果 Agent 需要访问数据库创建专用账号只授予所需表的 SELECT、INSERT 权限。如果 Agent 需要调用第三方 API使用单独的 API Key并设置配额限制。对 Agent 的工具调用日志做全量记录包括工具参数和执行结果便于安全审计。在原型里我对calculate工具做了字符白名单校验这就是一种安全收敛。生产环境上你可以使用专门的沙箱机制如 Docker 隔离、子进程执行来执行不可信代码。6.4 可观测性是 Agent OS 的生命线Agent 系统比普通 Web 系统更难排查问题因为它的执行路径往往跨越多次模型推理和多次工具调用。强烈建议从第一天就建立日志规范。每个任务至少要记录以下信息任务 ID 与父任务 ID用于追踪任务树Agent 名称输入消息和输出消息每次 LLM 调用的 token 开销工具调用的入参和返回值异常堆栈在原型里我们用了 Python 自带的logging这还不够。生产环境建议接入 OpenTelemetry 或 Langfuse 这类可观测平台将 Agent 的执行链路可视化。6.5 结果与状态管理要尽早设计原型里没有实现结果查询接口这是一个明显的短板。真实项目中任务提交后通常需要支持以下操作查询任务状态running / completed / failed查询任务执行结果取消一个任务重试一个失败的任务这些能力需要你设计一个TaskStore把任务的状态、输入、输出持久化到数据库。这也是 Agent OS 从“原型”走向“产品”的关键一步。7. 总结与下一步学习路线这篇文章从 Agent OS 的概念讲起解释了它为什么不是传统意义上的操作系统而是智能体运行时所需的一组基础设施抽象。然后我们通过一个完整的 Python 原型实现了 Agent 注册、任务调度、消息传递、记忆存储、工具调用等核心能力。通过这个实战项目你应该掌握了以下关键技能理解 Agent 的生命周期状态与状态管理方式。掌握 Agent 注册表模式能够在系统中统一管理多个 Agent。理解任务调度器的基本原理知道如何用asyncio.Queue实现简单的异步任务分发。掌握 Agent 工具调用的开发模式了解模型如何依据工具 Schema 生成调用指令。理解记忆模块的分层设计短期记忆与长期持久化各自的定位。下一步你可以从以下几个方向继续深入实现TaskStore支持任务结果查询与失败重试。将内存队列替换为 Redis Stream实现多实例部署。引入向量数据库为 Agent 添加语义检索长期记忆。研究多 Agent 编排模式例如“规划 Agent 拆解任务 → 执行 Agent 实际操作 → 汇总 Agent 输出”。学习 LangGraph、AutoGen、CrewAI 等成熟的 Agent 编排框架对照本文原型的模块理解成熟框架的设计取舍。如果这篇文章对你有帮助建议收藏备用后续你开发 Agent 应用时可以随时回来对照架构设计和代码思路。动手把原型跑起来才是掌握 Agent OS 核心思想的最好方式。