ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

Agent操作系统:从Harness平台到多Agent协作的工程实践

Agent操作系统:从Harness平台到多Agent协作的工程实践 1. 从跑个脚本到管一支队伍Agent 操作系统到底在解决什么如果你最近半年一直在折腾 AI Agent大概率经历过这样一个阶段一开始写个 Python 脚本调一次大模型 API拿回结果打印出来觉得挺爽。然后你想让它干点复杂的活——查资料、写代码、跑测试、改 bug、再跑一遍——于是你开始加循环、加判断、加工具调用。再往后你发现自己在写的东西已经不像一个脚本了更像一个调度中心谁先干、谁后干、失败了怎么重试、上下文怎么传递、状态存哪里、多个 Agent 之间怎么不打架。这个时候你其实已经在无意中造一个操作系统了。Agent 操作系统这个词听起来很唬人但拆开看一点都不玄。传统操作系统管的是什么管 CPU 时间片、管内存分配、管进程调度、管文件系统、管设备驱动。它的核心职责是把有限的硬件资源公平高效地分配给多个互相竞争的程序。而 Agent 操作系统管的是另一套资源大模型的调用次数token 预算、上下文窗口、工具权限、任务队列、以及多个 Agent 之间的协作关系。所以我在标题里用了未来的 Harness 平台这个说法。Harness 这个词在工程语境里本意是挽具、约束装置——它不产生动力但它把动力导向正确的方向。一个 Agent 的 Harness就是套在模型外面那一层负责约束、编排、观测、纠错的基础设施。模型本身是发动机Harness 是传动系统和方向盘。而Agent 操作系统则是把这套 Harness 从单个 Agent 的脚手架升级成多 Agent 的公共底座。这篇文章适合谁看三类人。第一类是做 Agent 开发、已经踩过上下文爆炸工具调用死循环多 Agent 状态错乱这些坑的工程师第二类是想理解 LLMOps 和 Agent 基础设施演进方向的技术负责人第三类是对操作系统概念有基础、想看看这套老思想怎么迁移到 AI 原生应用上的学习者。我会尽量把原理讲透同时给出可以直接抄的实操思路不玩虚的。需要先说明一点Agent 操作系统目前还没有一个像 Linux 那样公认的标准实现市面上更多是各家框架在往这个方向靠。所以下面讲的内容一部分是已经落地的工程实践一部分是基于操作系统经典原理和当前 Agent 框架现状做的合理推演我会明确区分哪些是实测有效、哪些是设计思路。2. 拆解 Agent 操作系统的四大核心子系统要理解一个操作系统最好的办法不是背定义而是看它管哪几件事。传统操作系统有进程管理、内存管理、文件系统、设备管理四大件。Agent 操作系统也有对应的四件套只是名字和内涵变了。这一章我逐个拆。2.1 进程管理Agent 的生命周期与调度在传统 OS 里进程是资源分配的基本单位有创建、就绪、运行、阻塞、终止五个状态。Agent 也一样。一个 Agent 从被创建拿到任务描述和初始上下文到就绪等待模型或工具返回、运行正在推理或执行工具、阻塞等待外部依赖比如等一个 API 响应或者等另一个 Agent 的结果、终止任务完成或失败整个生命周期需要被统一管理。这里最关键的问题是调度策略。传统 OS 用时间片轮转、优先级调度、多级反馈队列。Agent 的调度要考虑的东西更复杂一个任务该用哪个模型简单任务用便宜的小模型复杂推理用贵的大模型这就是成本感知调度。多个 Agent 抢同一个工具比如同一个数据库连接、同一个浏览器实例怎么办需要资源锁。一个 Agent 卡住了是等它还是超时杀掉重来需要超时与抢占机制。我实测下来最容易出问题的是阻塞态的处理。很多 Agent 框架在等工具返回时是同步阻塞的一个 Agent 卡住整个流程就停了。正确的做法是把 Agent 的执行做成事件驱动Agent 发起一个工具调用后立刻挂起把控制权交还给调度器等工具结果回来再唤醒它。这跟操作系统里的 I/O 多路复用是一个思路。2.2 内存管理上下文窗口就是你的物理内存这是我觉得整个类比里最精妙的一点。大模型的上下文窗口本质上就是 Agent 的物理内存——容量有限、访问有成本、超了就要换出。你不可能把所有历史对话、所有工具返回结果都塞进上下文就像你不可能把所有数据都塞进 RAM。传统 OS 有虚拟内存、分页、页面置换算法LRU、LFU。Agent 的上下文管理也需要类似机制分页把长对话切分成页每页是一个语义完整的片段比如一轮完整的工具调用结果。置换当上下文快满时把最不重要的页换出到外部存储向量数据库、文件需要时再换回来。压缩把多页内容用模型总结成一段摘要用摘要替代原文这是 Agent 特有的内存压缩。我踩过的一个坑是早期我图省事直接把所有工具返回的原始 JSON 全塞进上下文结果一个查数据库的操作返回了几万 token直接把窗口撑爆后面的推理全乱套。后来改成工具返回结果先经过一层摘要器只把关键字段和结论喂给主模型token 消耗直接降了 70% 以上。这个摘要器就是你的内存管理单元。2.3 文件系统Agent 的持久化记忆与状态Agent 不能只有短期记忆上下文还得有长期记忆持久化状态。传统 OS 用文件系统管理磁盘上的数据Agent 需要一套类似的记忆文件系统任务状态存哪、中间产物存哪、跨会话的知识存哪。常见的做法是分层存储层对应 OS 概念典型实现存什么上下文窗口寄存器/缓存模型 context当前推理必需的即时信息会话状态内存Redis / 进程内存当前任务的中间变量、工具调用记录长期记忆磁盘向量库 / 关系库跨会话的知识、用户偏好、历史结论归档冷存储对象存储完整日志、可回溯的执行轨迹这里有个经验不要把向量库当成万能记忆。向量检索适合模糊召回但任务状态这种需要精确读写的东西用关系库或者 KV 存储更靠谱。我见过有人把任务进度也塞进向量库结果检索出来的进度是旧的导致 Agent 重复劳动。记忆要分类型精确的走精确存储模糊的走向量。2.4 设备管理工具就是 Agent 的外设Agent 调用的每一个工具——搜索引擎、代码执行器、数据库、浏览器——本质上都是它的外设。传统 OS 通过设备驱动屏蔽硬件差异Agent 操作系统通过工具抽象层屏蔽工具差异。这一层的核心是统一的工具接口和权限控制。工具接口要统一成模型能理解的 schema通常是 JSON Schema这样模型不用关心底层是 HTTP 还是本地函数。权限控制则决定了哪个 Agent 能用哪个工具、能用到什么程度——比如一个负责查资料的 Agent 不应该有删数据库的权限。提示工具抽象层一定要做幂等性设计。Agent 因为重试机制同一个工具可能被调用多次。如果这个工具是扣款发邮件这种有副作用的操作重复调用就是事故。给每个工具调用加一个唯一 ID服务端做去重这是保命的设计。3. 为什么Harness比Framework更准确市面上大家习惯说Agent 框架Agent Framework但我在标题里特意用了 Harness。这两个词看着像内涵差别很大值得单独说清楚因为理解这个差别能帮你选对技术路线。3.1 框架是你按它的方式写Harness 是它围着你的 Agent 转框架Framework的特点是控制反转你填代码到它规定的槽位里它决定什么时候调用你。比如你写一个 LangChain 的 Chain你得按它的抽象来组织。框架越重你的自由度越低被绑定的风险越高。Harness 的思路反过来。它假设你的 Agent 核心逻辑是独立的、可替换的Harness 只是套在外面的一层壳负责观测、约束、编排、评测。你可以随时把里面的模型从 A 换成 B把 Agent 逻辑从手写换成别的实现Harness 不用大改。这个区别在实际项目里非常致命。我见过团队用某个重框架写了半年结果想换个模型供应商发现框架的抽象层把模型调用封死了改起来伤筋动骨。而 Harness 式的设计模型调用是一个可插拔的 adapter换供应商就是换个配置文件的事。3.2 Harness 工程的核心可观测性与可评测性Harness 这个词在 AI 圈火起来很大程度是因为评测Eval。你要评测一个 Agent 好不好不能只看它最后输出对不对还要看它中间走了哪些弯路、调了多少次工具、花了多少 token。这些都需要 Harness 来记录。一个合格的 Agent Harness 至少要能回答这些问题这次任务总共调了多少次模型每次的输入输出是什么工具调用了几次哪些成功了、哪些失败了、失败原因是什么整个执行轨迹trace长什么样哪一步是瓶颈同样的任务跑 10 次成功率是多少方差大不大没有这套观测你调 Agent 就是盲人摸象。我自己的习惯是任何 Agent 上线前先用 Harness 跑 50 次同样的任务看成功率和 token 分布。如果成功率低于 80%或者 token 消耗方差特别大说明有时走对了有时在瞎撞那就不能上线。3.3 Harness 和 Agent 的边界在哪里很多人搞不清 Harness 和 Agent 的区别。我的划分标准很简单Agent 负责做决策根据当前状态决定下一步调哪个工具、说什么话。Harness 负责管决策记录决策、约束决策、在决策出错时兜底、把多个 Agent 的决策协调起来。举个例子。Agent 决定我要调用搜索工具查一下 X。Harness 负责的是检查这个 Agent 有没有搜索权限、记录这次调用、设置超时、如果搜索失败按策略重试、把结果格式化后还给 Agent。Agent 只管我要查 XHarness 管怎么查、能不能查、查完怎么处理。这个边界划清楚之后你的系统会变得非常好维护。Agent 逻辑可以频繁迭代因为它是业务Harness 相对稳定因为它是基础设施。4. 手搓一个最小可用的 Agent 操作系统内核光讲概念没意思这一章我带你从零搭一个最小可用的 Agent 操作系统内核。不依赖任何重型框架用 Python 手写目的是让你看清每一层的职责。代码是示意性的重点是结构。4.1 定义核心数据结构Task、Agent、Tool、Context先把四个基础对象定义清楚。这是整个系统的地基。from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable import uuid import time class TaskState(Enum): PENDING pending # 就绪 RUNNING running # 运行 BLOCKED blocked # 阻塞等工具/等依赖 DONE done # 完成 FAILED failed # 失败 dataclass class Task: id: str field(default_factorylambda: str(uuid.uuid4())) goal: str # 任务目标描述 state: TaskState TaskState.PENDING parent_id: str | None None # 父任务支持任务树 created_at: float field(default_factorytime.time) result: Any None error: str | None None dataclass class Tool: name: str description: str schema: dict # JSON Schema给模型看的 func: Callable # 实际执行函数 idempotent: bool True # 是否幂等 timeout: float 30.0 dataclass class Context: Agent 的上下文对应 OS 的内存 messages: list field(default_factorylist) max_tokens: int 8000 # 超出预算时触发的压缩回调 compressor: Callable | None None这几个结构对应关系很清晰Task 是进程Tool 是设备Context 是内存。parent_id支持任务树这是后面多 Agent 协作的基础。4.2 调度器让 Agent 不再同步阻塞调度器是整个内核的心脏。它的职责是从就绪队列取任务、分配资源、执行、处理阻塞和唤醒。import asyncio from collections import deque class Scheduler: def __init__(self, max_concurrent: int 4): self.ready_queue deque() self.blocked: dict[str, asyncio.Future] {} self.running: dict[str, asyncio.Task] {} self.max_concurrent max_concurrent self.semaphore asyncio.Semaphore(max_concurrent) def submit(self, task: Task, runner: Callable): 提交任务到就绪队列 self.ready_queue.append((task, runner)) async def run(self): while self.ready_queue or self.running: # 有空闲槽位就启动新任务 while self.ready_queue and len(self.running) self.max_concurrent: task, runner self.ready_queue.popleft() task.state TaskState.RUNNING t asyncio.create_task(self._execute(task, runner)) self.running[task.id] t # 等待任意一个任务完成 if self.running: done, _ await asyncio.wait( self.running.values(), return_whenasyncio.FIRST_COMPLETED ) for t in done: for tid, rt in list(self.running.items()): if rt is t: del self.running[tid] async def _execute(self, task: Task, runner: Callable): async with self.semaphore: try: task.result await runner(task) task.state TaskState.DONE except Exception as e: task.error str(e) task.state TaskState.FAILED这段代码的关键在于用 asyncio 实现并发调度。max_concurrent就是你的CPU 核数——同时最多跑几个 Agent。超过这个数就排队。这跟操作系统的进程调度是一个道理资源有限必须排队。我实测下来max_concurrent的设置很有讲究。设太小吞吐上不去设太大模型 API 的速率限制rate limit会把你打爆而且多个 Agent 抢上下文资源容易乱。一般建议从 4 开始试根据 API 的 QPS 上限和任务平均耗时调整。4.3 上下文管理器token 预算的守门人上下文管理器负责在每次调用模型前检查 token 预算超了就压缩。这是防止内存溢出的关键。class ContextManager: def __init__(self, max_tokens: int 8000, reserve: int 1500): self.max_tokens max_tokens self.reserve reserve # 给模型输出预留的空间 def estimate_tokens(self, messages: list) - int: # 粗略估算中文约 1.5 字/token英文约 4 字符/token # 生产环境请用 tiktoken 等精确工具 total 0 for m in messages: content m.get(content, ) total len(content) // 2 10 return total async def prepare(self, ctx: Context, compressor: Callable) - list: budget self.max_tokens - self.reserve if self.estimate_tokens(ctx.messages) budget: return ctx.messages # 超预算触发压缩 compressed await compressor(ctx.messages, budget) ctx.messages compressed return compressed这里的reserve是个容易被忽略的细节。很多人只算输入 token忘了模型输出也要占窗口。如果你把输入塞到 7900输出空间只剩 100模型话还没说完就被截断了。留 1500 左右的输出空间是比较稳妥的。压缩策略我一般用滑动窗口 摘要保留最近 N 轮完整对话更早的内容用模型总结成一段话。这样既保住了近期上下文又不丢历史信息。4.4 工具网关统一入口 权限 幂等工具网关是所有工具调用的唯一入口。它做三件事权限校验、幂等去重、超时控制。class ToolGateway: def __init__(self): self.tools: dict[str, Tool] {} self.call_cache: dict[str, Any] {} # 幂等缓存 self.permissions: dict[str, set] {} # agent_id - 允许的工具集 def register(self, tool: Tool): self.tools[tool.name] tool def grant(self, agent_id: str, tool_names: list[str]): self.permissions[agent_id] set(tool_names) async def invoke(self, agent_id: str, tool_name: str, args: dict, call_id: str) - Any: # 1. 权限校验 if tool_name not in self.permissions.get(agent_id, set()): raise PermissionError(fAgent {agent_id} 无权调用 {tool_name}) tool self.tools[tool_name] # 2. 幂等去重 if tool.idempotent and call_id in self.call_cache: return self.call_cache[call_id] # 3. 超时执行 try: result await asyncio.wait_for( tool.func(**args), timeouttool.timeout ) except asyncio.TimeoutError: raise TimeoutError(f工具 {tool_name} 超时) if tool.idempotent: self.call_cache[call_id] result return resultcall_id由 Agent 在发起调用时生成重试时复用同一个 ID这样即使网络抖动导致重复请求服务端也只会真正执行一次。这个设计在处理支付、发消息这类有副作用的工具时是刚需。4.5 把内核跑起来一个完整的执行循环把上面几块拼起来就是一个最小的 Agent 执行循环async def agent_runner(task: Task): ctx Context() ctx.messages.append({role: user, content: task.goal}) while True: messages await ctx_mgr.prepare(ctx, compressor) # 调模型拿到下一步动作 action await call_llm(messages, toolstool_schemas) if action.type final: return action.content if action.type tool_call: result await gateway.invoke( agent_idtask.id, tool_nameaction.tool, argsaction.args, call_idaction.call_id ) ctx.messages.append({role: tool, content: str(result)})这个循环就是 Agent 的心跳。每一轮准备上下文 → 调模型 → 执行动作 → 更新上下文 → 再来一轮。直到模型给出最终答案。5. 多 Agent 协作从单进程到分布式系统单 Agent 跑通之后真正的挑战来了多个 Agent 怎么协作。这就像从单机程序进化到分布式系统复杂度是数量级上升的。5.1 三种协作拓扑主从、对等、流水线多 Agent 的协作结构我总结下来主要三种主从模式Orchestrator-Worker一个主 Agent 负责拆解任务、分配子任务多个从 Agent 负责执行。主 Agent 像项目经理从 Agent 像干活的。这种模式最好控制适合任务边界清晰的场景比如写一份报告拆成查资料写初稿审校。对等模式Peer-to-Peer多个 Agent 地位平等互相通信、协商。适合需要多视角讨论的场景比如辩论头脑风暴。但这种模式容易失控Agent 之间可能无限对话下去必须设置轮次上限。流水线模式PipelineAgent 按固定顺序串行前一个的输出是后一个的输入。适合流程固定的场景比如数据清洗 → 分析 → 可视化 → 报告。我实测下来主从模式是性价比最高的。对等模式看着酷但调试成本极高两个 Agent 互相甩锅或者陷入礼貌性循环的情况太常见了。除非你的场景真的需要多视角否则优先用主从。5.2 状态共享与隔离别让 Agent 互相踩脚多 Agent 最大的坑是状态污染。两个 Agent 同时读写同一份上下文结果互相覆盖任务就乱了。解决办法是状态隔离 显式共享。每个 Agent 有自己的私有上下文私有内存Agent 之间要共享信息必须通过显式的消息传递或共享黑板Blackboard。共享黑板是一个所有 Agent 都能读写的公共区域但读写要加锁。class Blackboard: def __init__(self): self.data {} self.locks {} async def write(self, key: str, value: Any, agent_id: str): if key not in self.locks: self.locks[key] asyncio.Lock() async with self.locks[key]: self.data[key] {value: value, writer: agent_id} async def read(self, key: str) - Any: return self.data.get(key, {}).get(value)这个设计跟操作系统的共享内存信号量是一个思路。共享是显式的同步是强制的这样才不会乱。5.3 死锁与活锁多 Agent 系统的经典病只要有多方竞争资源就有死锁。Agent A 等 Agent B 的结果Agent B 等 Agent A 的结果两个都卡住——这是死锁。Agent A 和 B 互相谦让谁都不肯先动手——这是活锁。防范死锁的经典手段在这里同样适用资源有序分配规定 Agent 获取资源的顺序避免循环等待。超时机制任何等待都设超时超时就放弃或重试。死锁检测定期扫描等待图发现环就打破牺牲一个 Agent 重来。活锁的防范更微妙核心是打破对称性。给每个 Agent 一个随机优先级或者规定当双方都想谦让时ID 小的先动手。听起来很土但实测有效。注意多 Agent 系统的调试难度是单 Agent 的十倍。我的建议是能用单 Agent 加工具解决的绝不上多 Agent。多 Agent 只在任务天然可并行、或者需要多角色视角时才用。为了架构好看而引入多 Agent是典型的过度设计。6. 落地时的真实坑我踩过的五个教训前面讲的都是应该怎么做这一章讲讲实际会怎么翻车。这些都是我在真实项目里踩出来的文档里不会写。6.1 坑一把 token 预算当儿戏账单教你做人我第一个上线的 Agent 项目没做 token 预算控制。结果有个任务因为工具返回了超大结果Agent 反复重试一晚上烧掉了几百块。后来我加了硬性预算每个任务有 token 上限超了就强制终止并报警。这个上限按任务类型设简单任务 1 万 token复杂任务 10 万绝不无限。6.2 坑二工具描述写得太随意模型乱调用工具的 description 是给模型看的写得好不好直接决定模型会不会用错。我早期写工具描述就一句话查询用户信息结果模型经常传错参数。后来改成详细描述参数含义、格式要求、返回什么、什么情况下用、什么情况下别用。改完之后工具调用准确率肉眼可见地提升。6.3 坑三没有 trace出问题只能靠猜Agent 出问题是常态关键是出问题后能不能快速定位。没有执行轨迹trace你只能看到任务失败了但不知道哪一步失败的。我现在的做法是每一步都记结构化日志时间戳、Agent ID、动作类型、输入、输出、耗时、token 消耗。出问题时把 trace 拉出来一看一目了然。6.4 坑四重试策略太粗暴副作用工具被重复执行前面提过幂等这里再强调一次。我见过最惨的事故是一个发通知的工具因为超时被重试了三次用户收到了三条一样的通知。修复方案就是给每次工具调用生成唯一 ID服务端按 ID 去重。这个改动很小但能救命。6.5 坑五评测集不更新Agent 悄悄退化Agent 依赖的模型会更新工具会变数据分布会漂移。如果你没有一套固定的评测集Agent 可能悄悄退化了你还不知道。我的习惯是维护一个 50-100 条的金标准任务集每次改动换模型、改 prompt、加工具都跑一遍看成功率有没有掉。这是 Agent 的回归测试。7. 从 Harness 到操作系统这条路会怎么走最后聊聊趋势但我不做空泛的展望只说几个我观察到的、正在发生的具体变化。第一个变化是Harness 正在从库变成平台。早期的 Harness 就是一个 Python 包你 import 进来用。现在越来越多的 Harness 开始提供 Web 界面、提供托管服务、提供团队协作能力。它正在变成一个你部署一次、全团队共用的基础设施这就是操作系统的雏形。第二个变化是评测和观测正在成为核心竞争力。模型能力会趋同工具会标准化真正拉开差距的是你对 Agent 行为的理解和控制能力。谁能把 trace 做得更细、把 eval 做得更准谁就能把 Agent 调得更稳。第三个变化是Agent 之间的互操作性。现在每个框架的 Agent 都是孤岛未来可能会出现类似Agent 通信协议的东西让不同框架的 Agent 能互相调用。这就像早期的计算机网络各家协议不通直到 TCP/IP 统一了标准。如果你现在要入局我的建议是别急着追新框架先把 Harness 这层的基本功练扎实。上下文管理、工具网关、trace 记录、eval 体系这四样东西不管未来框架怎么变都是刚需。把这四样做扎实了换任何框架你都能快速上手。我个人在实际操作中的体会是Agent 这个领域变化太快追热点不如打地基。操作系统那套几十年前就成熟的思想——进程调度、内存管理、资源隔离、死锁防范——在 Agent 时代几乎原封不动地复活了。理解了这些底层逻辑你看到任何新框架都能一眼看穿它在解决什么问题、用了什么老办法。这比记住十个框架的 API 有用得多。
返回列表