1. 为什么我们要把 Agent 的地基挖了重浇做 Agent 开发的人都有一个共同的体感原型跑通只要一个下午但想让它稳定扛住真实流量、接住复杂任务、还能让团队其他人接手迭代难度直接指数级上升。Orkas 这个项目我们做了快一年从最早的“能跑就行”到后来被各种边界情况按在地上摩擦最后下定决心做一次底层重构。这篇文章不讲虚的就聊我们为什么要把地基挖了重浇以及重浇的过程中哪些设计决策是真正经得起推敲的。先说背景。Orkas 是一个面向多智能体编排的 Agent 框架核心能力包括模型调用抽象、工具生态接入、多 Agent 协作调度、记忆管理这几块。早期版本用的是典型的“大单体”架构——一个 Orchestrator 类里塞了模型路由、工具注册、上下文拼装、状态管理所有逻辑。刚开始只有两三个 Agent 的时候没问题但当我们尝试接入十几个不同能力的 Agent、每个 Agent 又挂载十几个工具的时候整个系统开始出现各种诡异的问题上下文窗口爆炸、工具调用串线、Agent 之间状态互相污染、并发一上来就各种超时。最要命的是我们发现很多问题不是“加个 if else”能解决的而是架构层面的缺陷。比如模型调用层和编排层耦合太紧换一个模型供应商要改十几处代码工具注册是全局单例不同 Agent 想用同名但不同实现的工具直接冲突记忆存储没有分层短期记忆和长期记忆混在一起检索效率极低。这些问题在 demo 阶段看不出来但一旦上生产就是致命的。所以这次重构的核心目标很明确解耦、分层、可观测。解耦是指模型调用、工具执行、编排调度、记忆管理各层之间通过明确定义的接口通信任何一层的实现替换不影响其他层。分层是指把 Agent 的执行过程拆成清晰的阶段每个阶段有独立的生命周期和错误处理。可观测是指每一步执行都有完整的 trace 和 metric出问题能快速定位。这次重构不是小修小补而是把整个执行引擎重新设计了一遍。下面我会从架构设计、核心模块实现、实操踩坑几个维度展开把我们的经验完整分享出来。如果你正在做 Agent 框架选型或者正在从零搭建自己的 Agent 系统这些内容应该能帮你少走不少弯路。2. 重构前的架构诊断与核心问题拆解2.1 旧架构到底出了什么问题旧版 Orkas 的架构可以用一句话概括一个中心化的 Orchestrator 负责所有事情。用户请求进来后Orchestrator 先做意图识别然后决定调用哪个 Agent接着拼装上下文、调用模型、解析工具调用请求、执行工具、把结果塞回上下文、再次调用模型循环直到任务完成。这个流程在单 Agent 场景下没问题但多 Agent 场景下就暴露了三个致命缺陷。第一个缺陷是状态管理混乱。旧版把所有 Agent 的状态都存在一个全局的 Context 对象里不同 Agent 共享同一个上下文空间。这导致两个问题一是 Agent A 的中间结果会污染 Agent B 的推理过程二是当多个 Agent 并发执行时上下文读写没有隔离经常出现数据覆盖。我们曾经遇到过一个 caseAgent A 正在写一个文件Agent B 同时读取了同一个文件路径结果 B 读到了 A 写了一半的内容直接导致任务失败。第二个缺陷是工具调用没有沙箱。旧版工具注册是全局的所有 Agent 共享同一个工具池。这带来两个问题一是命名冲突两个 Agent 想用同一个名字但不同实现的工具时无法共存二是安全边界缺失一个 Agent 可以调用另一个 Agent 专属的工具导致权限越界。我们后来做安全审计的时候发现理论上一个被注入的恶意 Agent 可以调用文件系统工具删除任意文件这个风险太大了。第三个缺陷是模型调用层没有抽象。旧版直接在每个需要调模型的地方硬编码了 API 调用逻辑换模型供应商要改十几处代码而且不同模型的参数格式、返回结构、错误码都不一样维护成本极高。更麻烦的是我们想同时支持云端 API 和本地模型比如通过 LM Studio 跑的本地模型旧版架构根本做不到灵活切换。2.2 重构的核心设计原则针对上面这些问题我们定了四条设计原则。第一条接口隔离。模型调用、工具执行、记忆存储、编排调度四个核心能力各自定义独立的接口层与层之间只通过接口通信。这样任何一层的实现替换都不影响其他层。比如模型调用层定义了统一的ModelProvider接口云端 API 和本地模型各自实现这个接口上层编排逻辑完全无感知。第二条执行隔离。每个 Agent 的执行环境是独立的包括独立的上下文空间、独立的工具注册表、独立的记忆存储。Agent 之间通过消息传递通信不共享内存状态。这从根本上解决了状态污染和并发安全问题。第三条生命周期管理。Agent 的执行过程被拆成明确的阶段初始化、规划、执行、反思、终止。每个阶段有独立的生命周期钩子可以在任意阶段插入自定义逻辑也方便做精细化的错误处理和重试。第四条全链路可观测。每一步执行都生成结构化的 trace 数据包括输入输出、耗时、token 消耗、工具调用记录。这些数据统一收集到一个 TraceCollector 里支持实时查询和历史回溯。2.3 新旧架构对比维度旧架构新架构模型调用硬编码在各处统一 ModelProvider 接口支持多供应商工具注册全局单例Agent 级隔离支持命名空间状态管理全局共享 ContextAgent 级独立上下文消息传递通信并发模型无隔离靠锁执行单元隔离天然并发安全错误处理全局 try-catch分阶段错误处理支持重试和降级可观测性零散日志结构化 trace全链路可追溯扩展方式改 Orchestrator实现接口 注册插件这张表看起来简单但每一条背后都是血泪教训。比如“Agent 级隔离”这一条我们一开始觉得没必要觉得共享上下文更方便 Agent 之间交换信息。结果上线后第一个星期就遇到了状态污染导致的线上事故连夜改成隔离模式。所以有些设计原则真的是踩过坑才知道疼。3. 核心模块的重新设计与实现细节3.1 模型调用层统一抽象与多供应商适配模型调用层的重构是整个项目里最基础也最关键的一步。旧版代码里调用模型的地方散落在十几个文件里每个地方都直接写 HTTP 请求或者 SDK 调用。这种写法在只有一个模型供应商的时候还能忍但当我们想同时支持多个云端 API 和本地模型的时候就完全没法维护了。新架构里我们定义了一个ModelProvider接口核心方法只有三个class ModelProvider(ABC): abstractmethod async def chat(self, messages: List[Message], **kwargs) - ChatResponse: 发送对话请求返回模型响应 pass abstractmethod async def stream_chat(self, messages: List[Message], **kwargs) - AsyncIterator[ChatChunk]: 流式对话请求 pass abstractmethod def count_tokens(self, text: str) - int: 计算 token 数量 pass这个接口看起来简单但设计的时候有几个关键决策点。第一个决策点是消息格式的统一。不同模型供应商的消息格式不一样有的用rolecontent有的用systemuserassistant分开传有的支持多模态消息。我们定义了一个统一的Message结构包含role、content、name、tool_calls、tool_call_id这几个字段然后在各个 Provider 实现里做格式转换。这样上层编排逻辑只需要构造统一的Message对象不用关心底层是哪个供应商。第二个决策点是流式和非流式的统一。有些场景需要流式输出比如实时对话有些场景只需要最终结果比如批量任务。我们要求每个 Provider 同时实现chat和stream_chat两个方法上层根据需要选择。这里有个坑流式响应的错误处理和非流式完全不一样。非流式请求如果出错直接抛异常就行但流式请求可能已经返回了一部分内容然后中途出错这时候需要把已返回的内容和错误信息一起处理。我们的做法是在stream_chat里定义一个ChatChunk结构包含delta增量内容、finish_reason结束原因、error错误信息三个字段上层根据finish_reason判断是否正常结束。第三个决策点是重试和降级策略。模型调用失败是常态尤其是云端 API网络抖动、限流、超时都可能发生。我们在 Provider 层实现了统一的重试逻辑对于可重试的错误如超时、限流自动重试最多 3 次每次重试间隔指数退避对于不可重试的错误如认证失败、参数错误直接抛出。同时支持配置降级 Provider当主 Provider 连续失败超过阈值时自动切换到备用 Provider。class RetryableModelProvider(ModelProvider): def __init__(self, primary: ModelProvider, fallback: ModelProvider None, max_retries: int 3, base_delay: float 1.0): self.primary primary self.fallback fallback self.max_retries max_retries self.base_delay base_delay async def chat(self, messages, **kwargs): last_error None for attempt in range(self.max_retries): try: return await self.primary.chat(messages, **kwargs) except RetryableError as e: last_error e delay self.base_delay * (2 ** attempt) await asyncio.sleep(delay) except NonRetryableError: raise if self.fallback: return await self.fallback.chat(messages, **kwargs) raise last_error这个重试逻辑看起来简单但实际用起来有几个细节要注意。一是重试的时候要区分错误类型不是所有错误都值得重试。比如参数格式错误重试一万次也没用但超时重试一次可能就成功了。二是重试要有上限不然遇到持续故障会无限重试把资源耗尽。三是降级 Provider 的响应格式要和主 Provider 一致不然上层处理会出问题。3.2 工具生态从全局单例到 Agent 级隔离工具系统的重构是这次改动里最复杂的部分。旧版的工具注册是一个全局字典所有 Agent 共享。新架构里我们做了三个关键改动。第一个改动是工具注册表下沉到 Agent 级别。每个 Agent 实例拥有自己的工具注册表工具只在当前 Agent 内可见。这解决了命名冲突问题Agent A 可以有一个叫search的工具Agent B 也可以有一个叫search但实现完全不同的工具互不干扰。第二个改动是工具定义与实现分离。工具定义名称、描述、参数 schema和工具实现实际执行逻辑分开管理。定义部分用于生成模型可理解的工具描述实现部分用于实际执行。这样做的好处是同一个工具定义可以绑定不同的实现比如search工具在开发环境绑定 mock 实现在生产环境绑定真实搜索实现。dataclass class ToolDefinition: name: str description: str parameters: Dict[str, Any] # JSON Schema namespace: str default class ToolRegistry: def __init__(self): self._definitions: Dict[str, ToolDefinition] {} self._implementations: Dict[str, Callable] {} def register(self, definition: ToolDefinition, implementation: Callable): key f{definition.namespace}:{definition.name} self._definitions[key] definition self._implementations[key] implementation def get_specs(self) - List[Dict]: 生成模型可理解的工具描述列表 return [ { type: function, function: { name: d.name, description: d.description, parameters: d.parameters } } for d in self._definitions.values() ] async def execute(self, name: str, arguments: Dict) - Any: key fdefault:{name} if key not in self._implementations: raise ToolNotFoundError(fTool {name} not found) impl self._implementations[key] return await impl(**arguments)第三个改动是工具执行沙箱。每个工具执行都在独立的沙箱环境里运行有独立的超时控制、资源限制和错误隔离。工具执行失败不会影响 Agent 主流程只会返回一个错误结果让模型决定下一步。这里有个细节工具执行的超时时间要可配置不同工具的超时需求不一样。比如搜索工具可能需要 10 秒但计算器工具 1 秒就够了。我们在ToolDefinition里加了一个timeout字段默认 30 秒可以按工具单独配置。工具生态的另一个重要改动是支持工具组合。有些复杂操作需要多个工具配合完成比如“搜索并总结”需要先调用搜索工具再调用总结工具。旧版需要模型分两步调用新版支持定义一个组合工具内部自动编排多个子工具的执行。这个功能在需要固定流程的场景下特别有用可以减少模型调用次数提升执行效率。3.3 多智能体编排消息传递与执行隔离多 Agent 编排是 Orkas 的核心能力也是这次重构改动最大的部分。旧版的编排逻辑是“中心化调度”——一个 Orchestrator 决定所有 Agent 的执行顺序Agent 之间通过共享内存通信。新架构改成了“去中心化消息传递”——每个 Agent 是独立的执行单元Agent 之间通过消息队列通信。这个改动带来的最大好处是并发安全。旧版多个 Agent 并发执行时共享内存的读写需要加锁锁的粒度很难控制太粗影响性能太细容易死锁。新架构下每个 Agent 有独立的内存空间Agent 之间不共享任何可变状态天然并发安全。消息传递的设计有几个关键点。一是消息格式我们定义了统一的AgentMessage结构包含sender、receiver、content、message_type、correlation_id几个字段。message_type区分请求、响应、事件三种类型correlation_id用于关联请求和响应。二是消息路由每个 Agent 有一个唯一的 ID消息根据receiver字段路由到对应的 Agent。三是消息持久化所有消息都写入消息队列支持断点续传和审计回溯。dataclass class AgentMessage: sender: str receiver: str content: Any message_type: MessageType # REQUEST, RESPONSE, EVENT correlation_id: str field(default_factorylambda: str(uuid4())) timestamp: float field(default_factorytime.time) class MessageBus: def __init__(self): self._queues: Dict[str, asyncio.Queue] {} self._history: List[AgentMessage] [] def register(self, agent_id: str): self._queues[agent_id] asyncio.Queue() async def send(self, message: AgentMessage): self._history.append(message) if message.receiver in self._queues: await self._queues[message.receiver].put(message) else: raise AgentNotFoundError(message.receiver) async def receive(self, agent_id: str, timeout: float None) - AgentMessage: queue self._queues.get(agent_id) if not queue: raise AgentNotFoundError(agent_id) return await asyncio.wait_for(queue.get(), timeouttimeout)编排模式上我们支持三种顺序编排、并行编排、条件编排。顺序编排就是 Agent A 执行完传给 Agent B适合流水线场景。并行编排是多个 Agent 同时执行结果汇总后继续适合需要多角度分析的场景。条件编排是根据前一个 Agent 的输出决定下一个执行哪个 Agent适合分支决策场景。这里有个实操心得并行编排的时候要注意结果合并的顺序。如果多个 Agent 并行执行后需要合并结果合并顺序会影响最终输出。我们的做法是给每个并行分支分配一个优先级合并时按优先级排序。另外并行分支的数量要控制太多分支会导致资源竞争和上下文窗口爆炸。我们一般建议并行分支不超过 5 个。3.4 记忆管理分层存储与检索优化记忆管理是 Agent 框架里最容易被低估的部分。旧版的记忆存储就是一个简单的列表所有记忆混在一起检索的时候全量扫描。Agent 跑久了之后记忆列表越来越长检索效率直线下降而且上下文窗口很快就被塞满了。新架构里我们把记忆分成三层工作记忆、短期记忆、长期记忆。工作记忆是当前任务执行过程中的临时状态生命周期仅限于当前任务。比如 Agent 正在处理的文件内容、中间计算结果都存在工作记忆里。工作记忆的特点是读写频繁、容量小、任务结束即销毁。短期记忆是最近几次交互的历史记录生命周期是当前会话。比如用户最近说的几句话、Agent 最近几次的工具调用结果。短期记忆的特点是读写较频繁、容量中等、会话结束即归档。长期记忆是跨会话的持久化知识生命周期是永久。比如用户的偏好设置、历史任务的成功经验、领域知识。长期记忆的特点是读多写少、容量大、需要持久化存储。class MemoryManager: def __init__(self, working_capacity: int 50, short_term_capacity: int 200, long_term_store: VectorStore None): self.working deque(maxlenworking_capacity) self.short_term deque(maxlenshort_term_capacity) self.long_term long_term_store async def add(self, content: str, memory_type: MemoryType, metadata: Dict None): entry MemoryEntry(contentcontent, metadatametadata or {}, timestamptime.time()) if memory_type MemoryType.WORKING: self.working.append(entry) elif memory_type MemoryType.SHORT_TERM: self.short_term.append(entry) elif memory_type MemoryType.LONG_TERM: await self.long_term.add(entry) async def retrieve(self, query: str, top_k: int 5) - List[MemoryEntry]: results [] # 工作记忆全量返回容量小 results.extend(list(self.working)) # 短期记忆按时间倒序返回 results.extend(list(self.short_term)[-top_k:]) # 长期记忆向量检索 if self.long_term: results.extend(await self.long_term.search(query, top_ktop_k)) return results检索优化上我们做了两件事。一是分层检索不同层的记忆用不同的检索策略。工作记忆全量返回因为容量小短期记忆按时间倒序返回因为最近的交互最相关长期记忆用向量检索因为需要语义匹配。二是记忆压缩当短期记忆超过容量时把最旧的记忆压缩成摘要后存入长期记忆。压缩用一个小模型来做把多条记忆合并成一条摘要减少存储量同时保留关键信息。这里有个坑要提醒记忆压缩会丢失细节。我们曾经遇到过一个 caseAgent 在压缩记忆的时候把用户的一个关键约束条件压缩掉了导致后续任务执行偏离了用户预期。后来我们的做法是压缩前先做重要性评分重要的记忆不压缩直接存入长期记忆不重要的才压缩。重要性评分可以用规则比如包含数字、日期、否定词的记忆更重要或者用小模型来判断。4. 实操过程与核心环节实现4.1 从零搭建一个多 Agent 协作任务理论讲完了下面用一个实际例子把整个流程串起来。假设我们要搭建一个“市场调研报告生成”系统包含三个 Agent搜索 Agent 负责收集信息分析 Agent 负责提炼观点写作 Agent 负责生成报告。第一步是初始化框架和消息总线。from orkas import OrkasRuntime, MessageBus, ToolRegistry from orkas.providers import OpenAIProvider, LocalModelProvider # 初始化消息总线 bus MessageBus() # 初始化模型 Provider cloud_provider OpenAIProvider(api_keyyour-key, modelgpt-4) local_provider LocalModelProvider(base_urlhttp://localhost:1234/v1, modellocal-model) # 初始化运行时 runtime OrkasRuntime(message_busbus, default_providercloud_provider)第二步是定义和注册工具。搜索 Agent 需要搜索工具分析 Agent 需要数据统计工具写作 Agent 需要文件写入工具。# 搜索 Agent 的工具 search_registry ToolRegistry() search_registry.register( ToolDefinition( nameweb_search, description搜索互联网获取信息, parameters{ type: object, properties: { query: {type: string, description: 搜索关键词}, max_results: {type: integer, default: 5} }, required: [query] }, timeout15.0 ), web_search_impl ) # 分析 Agent 的工具 analysis_registry ToolRegistry() analysis_registry.register( ToolDefinition( namestatistics, description对数据进行统计分析, parameters{ type: object, properties: { data: {type: array, items: {type: number}}, operation: {type: string, enum: [mean, median, sum, count]} }, required: [data, operation] }, timeout5.0 ), statistics_impl ) # 写作 Agent 的工具 writer_registry ToolRegistry() writer_registry.register( ToolDefinition( namewrite_file, description将内容写入文件, parameters{ type: object, properties: { path: {type: string}, content: {type: string} }, required: [path, content] }, timeout10.0 ), write_file_impl )第三步是创建 Agent 实例并注册到运行时。# 创建搜索 Agent search_agent runtime.create_agent( agent_idsearcher, system_prompt你是一个信息搜索专家负责根据任务需求搜索相关信息。, toolssearch_registry, memory_config{working_capacity: 30, short_term_capacity: 100} ) # 创建分析 Agent analysis_agent runtime.create_agent( agent_idanalyzer, system_prompt你是一个数据分析专家负责对搜索到的信息进行提炼和分析。, toolsanalysis_registry, memory_config{working_capacity: 30, short_term_capacity: 100} ) # 创建写作 Agent writer_agent runtime.create_agent( agent_idwriter, system_prompt你是一个报告写作专家负责根据分析结果生成结构化的调研报告。, toolswriter_registry, memory_config{working_capacity: 30, short_term_capacity: 100} )第四步是定义编排流程。这里我们用顺序编排搜索 → 分析 → 写作。from orkas.orchestration import SequentialOrchestrator orchestrator SequentialOrchestrator( agents[search_agent, analysis_agent, writer_agent], message_busbus, timeout_per_agent120.0 ) # 执行任务 result await orchestrator.run( task调研 2026 年 AI Agent 框架的市场格局生成一份包含主要玩家、技术趋势、市场规模的报告, output_formatmarkdown )这个流程看起来简单但实际执行的时候有几个细节要注意。一是Agent 之间的消息格式要约定好。搜索 Agent 的输出是原始搜索结果列表分析 Agent 需要的是结构化的数据所以中间需要一个格式转换步骤。我们的做法是在 Agent 的 system prompt 里明确约定输出格式或者在编排层加一个转换函数。二是超时控制要分层。每个 Agent 有独立的超时时间整个编排流程也有总超时时间。单个 Agent 超时不影响其他 Agent但总超时到了整个流程终止。三是错误处理要分级。搜索 Agent 失败可以降级用缓存数据分析 Agent 失败可以跳过分析直接写作写作 Agent 失败则整个任务失败。4.2 并发场景下的性能调优Agent 系统扛并发是一个绕不开的话题。我们做过压测旧版架构在 50 并发的时候就开始出现超时和状态错误新版架构在 200 并发下依然稳定。这里分享几个关键的调优点。第一个调优点是连接池管理。模型 API 调用是 HTTP 请求每个请求都新建连接开销很大。我们用了aiohttp的连接池默认保持 100 个长连接复用连接减少握手开销。连接池的大小要根据并发量和 API 的限流策略来调整太小会导致请求排队太大会触发 API 限流。import aiohttp class OpenAIProvider(ModelProvider): def __init__(self, api_key: str, model: str, max_connections: int 100): self.api_key api_key self.model model self._session: aiohttp.ClientSession None self._max_connections max_connections async def _get_session(self) - aiohttp.ClientSession: if self._session is None or self._session.closed: connector aiohttp.TCPConnector( limitself._max_connections, limit_per_hostself._max_connections, ttl_dns_cache300 ) self._session aiohttp.ClientSession(connectorconnector) return self._session第二个调优点是批量请求合并。有些场景下多个 Agent 需要同时调用模型如果每个都单独发请求网络开销和 token 消耗都很大。我们的做法是在 Provider 层加一个批量接口把多个请求合并成一个批次发送。这个优化在 token 消耗上能省 20% 左右因为系统提示词只需要传一次。第三个调优点是上下文窗口管理。Agent 执行过程中上下文会越来越长如果不加控制很快就会超出模型的最大上下文长度。我们的做法是实时监控上下文长度超过阈值时触发压缩。压缩策略有两种一是滑动窗口保留最近 N 条消息二是摘要压缩把旧消息用模型总结成一条摘要。滑动窗口简单但会丢失信息摘要压缩保留信息但增加一次模型调用。我们一般建议混合使用最近 10 条消息保留原文更早的消息压缩成摘要。class ContextManager: def __init__(self, max_tokens: int 8000, keep_recent: int 10): self.max_tokens max_tokens self.keep_recent keep_recent async def manage(self, messages: List[Message], provider: ModelProvider) - List[Message]: total_tokens sum(provider.count_tokens(m.content) for m in messages) if total_tokens self.max_tokens: return messages # 保留最近 N 条 recent messages[-self.keep_recent:] older messages[:-self.keep_recent] # 旧消息压缩成摘要 if older: summary await self._summarize(older, provider) return [Message(rolesystem, contentf历史摘要{summary})] recent return recent async def _summarize(self, messages: List[Message], provider: ModelProvider) - str: text \n.join(f{m.role}: {m.content} for m in messages) response await provider.chat([ Message(rolesystem, content请将以下对话历史压缩成简洁的摘要保留关键信息。), Message(roleuser, contenttext) ]) return response.content第四个调优点是工具执行异步化。工具执行往往是 IO 密集型的比如搜索、文件读写同步执行会阻塞整个 Agent 流程。我们把所有工具执行都改成异步的多个工具可以并行执行。这里有个细节并行执行工具的时候要注意工具之间的依赖关系有依赖的工具必须串行执行。我们的做法是在工具定义里加一个depends_on字段编排器根据依赖关系自动决定执行顺序。4.3 本地模型与云端模型的混合调度Orkas 支持同时接入云端模型和本地模型这是很多团队的实际需求。云端模型能力强但成本高、有延迟本地模型成本低、响应快但能力有限。混合调度的核心思路是简单任务用本地模型复杂任务用云端模型。实现上我们定义了一个HybridProvider内部维护一个路由策略。路由策略可以基于规则比如根据任务类型、输入长度、历史成功率也可以基于模型用一个小模型来判断任务复杂度。class HybridProvider(ModelProvider): def __init__(self, local: ModelProvider, cloud: ModelProvider, complexity_threshold: int 500): self.local local self.cloud cloud self.complexity_threshold complexity_threshold async def chat(self, messages: List[Message], **kwargs) - ChatResponse: # 简单策略根据输入长度判断 total_length sum(len(m.content) for m in messages) if total_length self.complexity_threshold: try: return await self.local.chat(messages, **kwargs) except Exception: # 本地模型失败降级到云端 return await self.cloud.chat(messages, **kwargs) else: return await self.cloud.chat(messages, **kwargs)这个策略看起来简单但实际用起来效果不错。我们的测试数据显示在客服问答场景下70% 的请求可以用本地模型处理只有 30% 的复杂请求需要云端模型整体成本降低了 60% 以上。这里有个坑要提醒本地模型和云端模型的输出格式可能不一致。本地模型尤其是小模型在工具调用格式上经常出错比如 JSON 格式不对、参数类型错误。我们的做法是在 Provider 层加一个输出校验和修复逻辑对本地模型的输出做格式校验不合法的话自动重试或者降级到云端模型。5. 常见问题与排查技巧实录5.1 工具调用串线问题问题现象多个 Agent 并发执行时Agent A 的工具调用结果返回给了 Agent B。排查思路首先检查工具注册表是否隔离。旧版全局注册表是常见原因新版每个 Agent 独立注册表一般不会出现这个问题。如果注册表已经隔离检查消息总线的路由逻辑确认receiver字段是否正确。最后检查工具执行的上下文传递确认correlation_id是否贯穿整个调用链。解决方案确保每个工具调用都携带correlation_id工具执行结果根据correlation_id路由回正确的 Agent。我们在ToolRegistry.execute方法里强制要求传入correlation_id执行结果通过消息总线发回。5.2 上下文窗口爆炸问题现象Agent 执行到一半突然报错“context length exceeded”。排查思路检查上下文管理器的配置确认max_tokens是否设置合理。检查是否有工具返回了超大结果比如搜索返回了完整网页内容。检查记忆检索是否返回了过多条目。解决方案设置合理的max_tokens一般建议模型最大上下文的 70%工具返回结果做截断处理比如搜索结果只保留摘要记忆检索限制top_k数量。另外可以在 Agent 的 system prompt 里明确要求模型精简输出。5.3 模型调用超时问题现象模型调用频繁超时尤其是高峰期。排查思路检查网络连接和 API 端点延迟。检查连接池配置是否合理。检查是否有大量并发请求同时发出。解决方案增加连接池大小配置合理的超时时间一般建议 30-60 秒实现请求队列和限流。对于超时请求配置自动重试和降级策略。5.4 常见问题速查表问题可能原因排查方法解决方案工具调用串线注册表未隔离/路由错误检查注册表和消息路由隔离注册表携带 correlation_id上下文爆炸未压缩/工具返回过大检查上下文长度和工具输出压缩上下文截断工具输出模型超时网络延迟/并发过高检查网络和并发量增加连接池配置重试降级状态污染共享内存/无隔离检查 Agent 状态存储Agent 级隔离消息传递通信记忆检索慢全量扫描/无索引检查记忆存储结构分层存储向量检索本地模型格式错误模型能力不足检查输出格式格式校验自动降级5.5 独家避坑技巧第一个技巧是给每个 Agent 加执行预算。Agent 执行过程中可能陷入死循环比如反复调用同一个工具设置一个最大执行步数或最大 token 消耗超过预算强制终止。我们的默认预算是 50 步或 100k token可以根据任务复杂度调整。第二个技巧是工具执行结果做缓存。同一个工具用相同参数调用多次结果应该是一样的幂等工具。我们加了一个工具结果缓存相同参数的调用直接返回缓存结果减少重复执行。这个优化在搜索类工具上效果特别明显能减少 40% 的重复调用。第三个技巧是Agent 之间通信加超时。Agent A 给 Agent B 发消息后如果 B 在超时时间内没有响应A 应该继续执行而不是无限等待。我们的做法是在消息总线上加超时控制超时后返回一个默认响应或者错误让调用方决定下一步。第四个技巧是定期做全链路压测。Agent 系统的性能瓶颈往往不在单个模块而在模块之间的交互。我们每个月做一次全链路压测模拟真实流量模式找出瓶颈点。压测的时候要注意模拟真实的工具调用延迟和模型响应延迟不然压测结果会偏乐观。6. 重构后的效果与后续演进方向重构完成后我们做了一轮完整的对比测试。在相同的硬件环境和任务集下新版 Orkas 的吞吐量提升了 3 倍平均响应延迟降低了 40%错误率从 5% 降到了 0.5% 以下。更重要的是代码的可维护性大幅提升新加一个 Agent 只需要实现接口和注册工具不用改核心逻辑。从架构层面看这次重构最大的价值是建立了清晰的扩展点。模型调用层可以接入任何模型供应商工具层可以注册任何工具编排层可以定义任何编排模式记忆层可以替换任何存储后端。这种插件化的设计让 Orkas 从一个“能用的框架”变成了一个“可演进的平台”。后续我们计划在几个方向继续演进。一是Agent 能力评估建立一套标准化的评估体系量化每个 Agent 的能力边界和可靠性。二是自适应编排根据任务特征自动选择最优的编排模式和 Agent 组合。三是安全增强在工具执行沙箱的基础上增加更细粒度的权限控制和审计日志。如果你也在做 Agent 框架的开发我的建议是不要等到问题堆积如山才重构在架构还清晰的时候就把分层和隔离做好。前期多花一周设计接口后期能省一个月修 bug。另外可观测性一定要从第一天就做没有 trace 的 Agent 系统就像没有日志的后端服务出了问题只能靠猜。 多智能体框架Orkas底层重构：解耦、分层与可观测的工程实践
1. 为什么我们要把 Agent 的地基挖了重浇做 Agent 开发的人都有一个共同的体感原型跑通只要一个下午但想让它稳定扛住真实流量、接住复杂任务、还能让团队其他人接手迭代难度直接指数级上升。Orkas 这个项目我们做了快一年从最早的“能跑就行”到后来被各种边界情况按在地上摩擦最后下定决心做一次底层重构。这篇文章不讲虚的就聊我们为什么要把地基挖了重浇以及重浇的过程中哪些设计决策是真正经得起推敲的。先说背景。Orkas 是一个面向多智能体编排的 Agent 框架核心能力包括模型调用抽象、工具生态接入、多 Agent 协作调度、记忆管理这几块。早期版本用的是典型的“大单体”架构——一个 Orchestrator 类里塞了模型路由、工具注册、上下文拼装、状态管理所有逻辑。刚开始只有两三个 Agent 的时候没问题但当我们尝试接入十几个不同能力的 Agent、每个 Agent 又挂载十几个工具的时候整个系统开始出现各种诡异的问题上下文窗口爆炸、工具调用串线、Agent 之间状态互相污染、并发一上来就各种超时。最要命的是我们发现很多问题不是“加个 if else”能解决的而是架构层面的缺陷。比如模型调用层和编排层耦合太紧换一个模型供应商要改十几处代码工具注册是全局单例不同 Agent 想用同名但不同实现的工具直接冲突记忆存储没有分层短期记忆和长期记忆混在一起检索效率极低。这些问题在 demo 阶段看不出来但一旦上生产就是致命的。所以这次重构的核心目标很明确解耦、分层、可观测。解耦是指模型调用、工具执行、编排调度、记忆管理各层之间通过明确定义的接口通信任何一层的实现替换不影响其他层。分层是指把 Agent 的执行过程拆成清晰的阶段每个阶段有独立的生命周期和错误处理。可观测是指每一步执行都有完整的 trace 和 metric出问题能快速定位。这次重构不是小修小补而是把整个执行引擎重新设计了一遍。下面我会从架构设计、核心模块实现、实操踩坑几个维度展开把我们的经验完整分享出来。如果你正在做 Agent 框架选型或者正在从零搭建自己的 Agent 系统这些内容应该能帮你少走不少弯路。2. 重构前的架构诊断与核心问题拆解2.1 旧架构到底出了什么问题旧版 Orkas 的架构可以用一句话概括一个中心化的 Orchestrator 负责所有事情。用户请求进来后Orchestrator 先做意图识别然后决定调用哪个 Agent接着拼装上下文、调用模型、解析工具调用请求、执行工具、把结果塞回上下文、再次调用模型循环直到任务完成。这个流程在单 Agent 场景下没问题但多 Agent 场景下就暴露了三个致命缺陷。第一个缺陷是状态管理混乱。旧版把所有 Agent 的状态都存在一个全局的 Context 对象里不同 Agent 共享同一个上下文空间。这导致两个问题一是 Agent A 的中间结果会污染 Agent B 的推理过程二是当多个 Agent 并发执行时上下文读写没有隔离经常出现数据覆盖。我们曾经遇到过一个 caseAgent A 正在写一个文件Agent B 同时读取了同一个文件路径结果 B 读到了 A 写了一半的内容直接导致任务失败。第二个缺陷是工具调用没有沙箱。旧版工具注册是全局的所有 Agent 共享同一个工具池。这带来两个问题一是命名冲突两个 Agent 想用同一个名字但不同实现的工具时无法共存二是安全边界缺失一个 Agent 可以调用另一个 Agent 专属的工具导致权限越界。我们后来做安全审计的时候发现理论上一个被注入的恶意 Agent 可以调用文件系统工具删除任意文件这个风险太大了。第三个缺陷是模型调用层没有抽象。旧版直接在每个需要调模型的地方硬编码了 API 调用逻辑换模型供应商要改十几处代码而且不同模型的参数格式、返回结构、错误码都不一样维护成本极高。更麻烦的是我们想同时支持云端 API 和本地模型比如通过 LM Studio 跑的本地模型旧版架构根本做不到灵活切换。2.2 重构的核心设计原则针对上面这些问题我们定了四条设计原则。第一条接口隔离。模型调用、工具执行、记忆存储、编排调度四个核心能力各自定义独立的接口层与层之间只通过接口通信。这样任何一层的实现替换都不影响其他层。比如模型调用层定义了统一的ModelProvider接口云端 API 和本地模型各自实现这个接口上层编排逻辑完全无感知。第二条执行隔离。每个 Agent 的执行环境是独立的包括独立的上下文空间、独立的工具注册表、独立的记忆存储。Agent 之间通过消息传递通信不共享内存状态。这从根本上解决了状态污染和并发安全问题。第三条生命周期管理。Agent 的执行过程被拆成明确的阶段初始化、规划、执行、反思、终止。每个阶段有独立的生命周期钩子可以在任意阶段插入自定义逻辑也方便做精细化的错误处理和重试。第四条全链路可观测。每一步执行都生成结构化的 trace 数据包括输入输出、耗时、token 消耗、工具调用记录。这些数据统一收集到一个 TraceCollector 里支持实时查询和历史回溯。2.3 新旧架构对比维度旧架构新架构模型调用硬编码在各处统一 ModelProvider 接口支持多供应商工具注册全局单例Agent 级隔离支持命名空间状态管理全局共享 ContextAgent 级独立上下文消息传递通信并发模型无隔离靠锁执行单元隔离天然并发安全错误处理全局 try-catch分阶段错误处理支持重试和降级可观测性零散日志结构化 trace全链路可追溯扩展方式改 Orchestrator实现接口 注册插件这张表看起来简单但每一条背后都是血泪教训。比如“Agent 级隔离”这一条我们一开始觉得没必要觉得共享上下文更方便 Agent 之间交换信息。结果上线后第一个星期就遇到了状态污染导致的线上事故连夜改成隔离模式。所以有些设计原则真的是踩过坑才知道疼。3. 核心模块的重新设计与实现细节3.1 模型调用层统一抽象与多供应商适配模型调用层的重构是整个项目里最基础也最关键的一步。旧版代码里调用模型的地方散落在十几个文件里每个地方都直接写 HTTP 请求或者 SDK 调用。这种写法在只有一个模型供应商的时候还能忍但当我们想同时支持多个云端 API 和本地模型的时候就完全没法维护了。新架构里我们定义了一个ModelProvider接口核心方法只有三个class ModelProvider(ABC): abstractmethod async def chat(self, messages: List[Message], **kwargs) - ChatResponse: 发送对话请求返回模型响应 pass abstractmethod async def stream_chat(self, messages: List[Message], **kwargs) - AsyncIterator[ChatChunk]: 流式对话请求 pass abstractmethod def count_tokens(self, text: str) - int: 计算 token 数量 pass这个接口看起来简单但设计的时候有几个关键决策点。第一个决策点是消息格式的统一。不同模型供应商的消息格式不一样有的用rolecontent有的用systemuserassistant分开传有的支持多模态消息。我们定义了一个统一的Message结构包含role、content、name、tool_calls、tool_call_id这几个字段然后在各个 Provider 实现里做格式转换。这样上层编排逻辑只需要构造统一的Message对象不用关心底层是哪个供应商。第二个决策点是流式和非流式的统一。有些场景需要流式输出比如实时对话有些场景只需要最终结果比如批量任务。我们要求每个 Provider 同时实现chat和stream_chat两个方法上层根据需要选择。这里有个坑流式响应的错误处理和非流式完全不一样。非流式请求如果出错直接抛异常就行但流式请求可能已经返回了一部分内容然后中途出错这时候需要把已返回的内容和错误信息一起处理。我们的做法是在stream_chat里定义一个ChatChunk结构包含delta增量内容、finish_reason结束原因、error错误信息三个字段上层根据finish_reason判断是否正常结束。第三个决策点是重试和降级策略。模型调用失败是常态尤其是云端 API网络抖动、限流、超时都可能发生。我们在 Provider 层实现了统一的重试逻辑对于可重试的错误如超时、限流自动重试最多 3 次每次重试间隔指数退避对于不可重试的错误如认证失败、参数错误直接抛出。同时支持配置降级 Provider当主 Provider 连续失败超过阈值时自动切换到备用 Provider。class RetryableModelProvider(ModelProvider): def __init__(self, primary: ModelProvider, fallback: ModelProvider None, max_retries: int 3, base_delay: float 1.0): self.primary primary self.fallback fallback self.max_retries max_retries self.base_delay base_delay async def chat(self, messages, **kwargs): last_error None for attempt in range(self.max_retries): try: return await self.primary.chat(messages, **kwargs) except RetryableError as e: last_error e delay self.base_delay * (2 ** attempt) await asyncio.sleep(delay) except NonRetryableError: raise if self.fallback: return await self.fallback.chat(messages, **kwargs) raise last_error这个重试逻辑看起来简单但实际用起来有几个细节要注意。一是重试的时候要区分错误类型不是所有错误都值得重试。比如参数格式错误重试一万次也没用但超时重试一次可能就成功了。二是重试要有上限不然遇到持续故障会无限重试把资源耗尽。三是降级 Provider 的响应格式要和主 Provider 一致不然上层处理会出问题。3.2 工具生态从全局单例到 Agent 级隔离工具系统的重构是这次改动里最复杂的部分。旧版的工具注册是一个全局字典所有 Agent 共享。新架构里我们做了三个关键改动。第一个改动是工具注册表下沉到 Agent 级别。每个 Agent 实例拥有自己的工具注册表工具只在当前 Agent 内可见。这解决了命名冲突问题Agent A 可以有一个叫search的工具Agent B 也可以有一个叫search但实现完全不同的工具互不干扰。第二个改动是工具定义与实现分离。工具定义名称、描述、参数 schema和工具实现实际执行逻辑分开管理。定义部分用于生成模型可理解的工具描述实现部分用于实际执行。这样做的好处是同一个工具定义可以绑定不同的实现比如search工具在开发环境绑定 mock 实现在生产环境绑定真实搜索实现。dataclass class ToolDefinition: name: str description: str parameters: Dict[str, Any] # JSON Schema namespace: str default class ToolRegistry: def __init__(self): self._definitions: Dict[str, ToolDefinition] {} self._implementations: Dict[str, Callable] {} def register(self, definition: ToolDefinition, implementation: Callable): key f{definition.namespace}:{definition.name} self._definitions[key] definition self._implementations[key] implementation def get_specs(self) - List[Dict]: 生成模型可理解的工具描述列表 return [ { type: function, function: { name: d.name, description: d.description, parameters: d.parameters } } for d in self._definitions.values() ] async def execute(self, name: str, arguments: Dict) - Any: key fdefault:{name} if key not in self._implementations: raise ToolNotFoundError(fTool {name} not found) impl self._implementations[key] return await impl(**arguments)第三个改动是工具执行沙箱。每个工具执行都在独立的沙箱环境里运行有独立的超时控制、资源限制和错误隔离。工具执行失败不会影响 Agent 主流程只会返回一个错误结果让模型决定下一步。这里有个细节工具执行的超时时间要可配置不同工具的超时需求不一样。比如搜索工具可能需要 10 秒但计算器工具 1 秒就够了。我们在ToolDefinition里加了一个timeout字段默认 30 秒可以按工具单独配置。工具生态的另一个重要改动是支持工具组合。有些复杂操作需要多个工具配合完成比如“搜索并总结”需要先调用搜索工具再调用总结工具。旧版需要模型分两步调用新版支持定义一个组合工具内部自动编排多个子工具的执行。这个功能在需要固定流程的场景下特别有用可以减少模型调用次数提升执行效率。3.3 多智能体编排消息传递与执行隔离多 Agent 编排是 Orkas 的核心能力也是这次重构改动最大的部分。旧版的编排逻辑是“中心化调度”——一个 Orchestrator 决定所有 Agent 的执行顺序Agent 之间通过共享内存通信。新架构改成了“去中心化消息传递”——每个 Agent 是独立的执行单元Agent 之间通过消息队列通信。这个改动带来的最大好处是并发安全。旧版多个 Agent 并发执行时共享内存的读写需要加锁锁的粒度很难控制太粗影响性能太细容易死锁。新架构下每个 Agent 有独立的内存空间Agent 之间不共享任何可变状态天然并发安全。消息传递的设计有几个关键点。一是消息格式我们定义了统一的AgentMessage结构包含sender、receiver、content、message_type、correlation_id几个字段。message_type区分请求、响应、事件三种类型correlation_id用于关联请求和响应。二是消息路由每个 Agent 有一个唯一的 ID消息根据receiver字段路由到对应的 Agent。三是消息持久化所有消息都写入消息队列支持断点续传和审计回溯。dataclass class AgentMessage: sender: str receiver: str content: Any message_type: MessageType # REQUEST, RESPONSE, EVENT correlation_id: str field(default_factorylambda: str(uuid4())) timestamp: float field(default_factorytime.time) class MessageBus: def __init__(self): self._queues: Dict[str, asyncio.Queue] {} self._history: List[AgentMessage] [] def register(self, agent_id: str): self._queues[agent_id] asyncio.Queue() async def send(self, message: AgentMessage): self._history.append(message) if message.receiver in self._queues: await self._queues[message.receiver].put(message) else: raise AgentNotFoundError(message.receiver) async def receive(self, agent_id: str, timeout: float None) - AgentMessage: queue self._queues.get(agent_id) if not queue: raise AgentNotFoundError(agent_id) return await asyncio.wait_for(queue.get(), timeouttimeout)编排模式上我们支持三种顺序编排、并行编排、条件编排。顺序编排就是 Agent A 执行完传给 Agent B适合流水线场景。并行编排是多个 Agent 同时执行结果汇总后继续适合需要多角度分析的场景。条件编排是根据前一个 Agent 的输出决定下一个执行哪个 Agent适合分支决策场景。这里有个实操心得并行编排的时候要注意结果合并的顺序。如果多个 Agent 并行执行后需要合并结果合并顺序会影响最终输出。我们的做法是给每个并行分支分配一个优先级合并时按优先级排序。另外并行分支的数量要控制太多分支会导致资源竞争和上下文窗口爆炸。我们一般建议并行分支不超过 5 个。3.4 记忆管理分层存储与检索优化记忆管理是 Agent 框架里最容易被低估的部分。旧版的记忆存储就是一个简单的列表所有记忆混在一起检索的时候全量扫描。Agent 跑久了之后记忆列表越来越长检索效率直线下降而且上下文窗口很快就被塞满了。新架构里我们把记忆分成三层工作记忆、短期记忆、长期记忆。工作记忆是当前任务执行过程中的临时状态生命周期仅限于当前任务。比如 Agent 正在处理的文件内容、中间计算结果都存在工作记忆里。工作记忆的特点是读写频繁、容量小、任务结束即销毁。短期记忆是最近几次交互的历史记录生命周期是当前会话。比如用户最近说的几句话、Agent 最近几次的工具调用结果。短期记忆的特点是读写较频繁、容量中等、会话结束即归档。长期记忆是跨会话的持久化知识生命周期是永久。比如用户的偏好设置、历史任务的成功经验、领域知识。长期记忆的特点是读多写少、容量大、需要持久化存储。class MemoryManager: def __init__(self, working_capacity: int 50, short_term_capacity: int 200, long_term_store: VectorStore None): self.working deque(maxlenworking_capacity) self.short_term deque(maxlenshort_term_capacity) self.long_term long_term_store async def add(self, content: str, memory_type: MemoryType, metadata: Dict None): entry MemoryEntry(contentcontent, metadatametadata or {}, timestamptime.time()) if memory_type MemoryType.WORKING: self.working.append(entry) elif memory_type MemoryType.SHORT_TERM: self.short_term.append(entry) elif memory_type MemoryType.LONG_TERM: await self.long_term.add(entry) async def retrieve(self, query: str, top_k: int 5) - List[MemoryEntry]: results [] # 工作记忆全量返回容量小 results.extend(list(self.working)) # 短期记忆按时间倒序返回 results.extend(list(self.short_term)[-top_k:]) # 长期记忆向量检索 if self.long_term: results.extend(await self.long_term.search(query, top_ktop_k)) return results检索优化上我们做了两件事。一是分层检索不同层的记忆用不同的检索策略。工作记忆全量返回因为容量小短期记忆按时间倒序返回因为最近的交互最相关长期记忆用向量检索因为需要语义匹配。二是记忆压缩当短期记忆超过容量时把最旧的记忆压缩成摘要后存入长期记忆。压缩用一个小模型来做把多条记忆合并成一条摘要减少存储量同时保留关键信息。这里有个坑要提醒记忆压缩会丢失细节。我们曾经遇到过一个 caseAgent 在压缩记忆的时候把用户的一个关键约束条件压缩掉了导致后续任务执行偏离了用户预期。后来我们的做法是压缩前先做重要性评分重要的记忆不压缩直接存入长期记忆不重要的才压缩。重要性评分可以用规则比如包含数字、日期、否定词的记忆更重要或者用小模型来判断。4. 实操过程与核心环节实现4.1 从零搭建一个多 Agent 协作任务理论讲完了下面用一个实际例子把整个流程串起来。假设我们要搭建一个“市场调研报告生成”系统包含三个 Agent搜索 Agent 负责收集信息分析 Agent 负责提炼观点写作 Agent 负责生成报告。第一步是初始化框架和消息总线。from orkas import OrkasRuntime, MessageBus, ToolRegistry from orkas.providers import OpenAIProvider, LocalModelProvider # 初始化消息总线 bus MessageBus() # 初始化模型 Provider cloud_provider OpenAIProvider(api_keyyour-key, modelgpt-4) local_provider LocalModelProvider(base_urlhttp://localhost:1234/v1, modellocal-model) # 初始化运行时 runtime OrkasRuntime(message_busbus, default_providercloud_provider)第二步是定义和注册工具。搜索 Agent 需要搜索工具分析 Agent 需要数据统计工具写作 Agent 需要文件写入工具。# 搜索 Agent 的工具 search_registry ToolRegistry() search_registry.register( ToolDefinition( nameweb_search, description搜索互联网获取信息, parameters{ type: object, properties: { query: {type: string, description: 搜索关键词}, max_results: {type: integer, default: 5} }, required: [query] }, timeout15.0 ), web_search_impl ) # 分析 Agent 的工具 analysis_registry ToolRegistry() analysis_registry.register( ToolDefinition( namestatistics, description对数据进行统计分析, parameters{ type: object, properties: { data: {type: array, items: {type: number}}, operation: {type: string, enum: [mean, median, sum, count]} }, required: [data, operation] }, timeout5.0 ), statistics_impl ) # 写作 Agent 的工具 writer_registry ToolRegistry() writer_registry.register( ToolDefinition( namewrite_file, description将内容写入文件, parameters{ type: object, properties: { path: {type: string}, content: {type: string} }, required: [path, content] }, timeout10.0 ), write_file_impl )第三步是创建 Agent 实例并注册到运行时。# 创建搜索 Agent search_agent runtime.create_agent( agent_idsearcher, system_prompt你是一个信息搜索专家负责根据任务需求搜索相关信息。, toolssearch_registry, memory_config{working_capacity: 30, short_term_capacity: 100} ) # 创建分析 Agent analysis_agent runtime.create_agent( agent_idanalyzer, system_prompt你是一个数据分析专家负责对搜索到的信息进行提炼和分析。, toolsanalysis_registry, memory_config{working_capacity: 30, short_term_capacity: 100} ) # 创建写作 Agent writer_agent runtime.create_agent( agent_idwriter, system_prompt你是一个报告写作专家负责根据分析结果生成结构化的调研报告。, toolswriter_registry, memory_config{working_capacity: 30, short_term_capacity: 100} )第四步是定义编排流程。这里我们用顺序编排搜索 → 分析 → 写作。from orkas.orchestration import SequentialOrchestrator orchestrator SequentialOrchestrator( agents[search_agent, analysis_agent, writer_agent], message_busbus, timeout_per_agent120.0 ) # 执行任务 result await orchestrator.run( task调研 2026 年 AI Agent 框架的市场格局生成一份包含主要玩家、技术趋势、市场规模的报告, output_formatmarkdown )这个流程看起来简单但实际执行的时候有几个细节要注意。一是Agent 之间的消息格式要约定好。搜索 Agent 的输出是原始搜索结果列表分析 Agent 需要的是结构化的数据所以中间需要一个格式转换步骤。我们的做法是在 Agent 的 system prompt 里明确约定输出格式或者在编排层加一个转换函数。二是超时控制要分层。每个 Agent 有独立的超时时间整个编排流程也有总超时时间。单个 Agent 超时不影响其他 Agent但总超时到了整个流程终止。三是错误处理要分级。搜索 Agent 失败可以降级用缓存数据分析 Agent 失败可以跳过分析直接写作写作 Agent 失败则整个任务失败。4.2 并发场景下的性能调优Agent 系统扛并发是一个绕不开的话题。我们做过压测旧版架构在 50 并发的时候就开始出现超时和状态错误新版架构在 200 并发下依然稳定。这里分享几个关键的调优点。第一个调优点是连接池管理。模型 API 调用是 HTTP 请求每个请求都新建连接开销很大。我们用了aiohttp的连接池默认保持 100 个长连接复用连接减少握手开销。连接池的大小要根据并发量和 API 的限流策略来调整太小会导致请求排队太大会触发 API 限流。import aiohttp class OpenAIProvider(ModelProvider): def __init__(self, api_key: str, model: str, max_connections: int 100): self.api_key api_key self.model model self._session: aiohttp.ClientSession None self._max_connections max_connections async def _get_session(self) - aiohttp.ClientSession: if self._session is None or self._session.closed: connector aiohttp.TCPConnector( limitself._max_connections, limit_per_hostself._max_connections, ttl_dns_cache300 ) self._session aiohttp.ClientSession(connectorconnector) return self._session第二个调优点是批量请求合并。有些场景下多个 Agent 需要同时调用模型如果每个都单独发请求网络开销和 token 消耗都很大。我们的做法是在 Provider 层加一个批量接口把多个请求合并成一个批次发送。这个优化在 token 消耗上能省 20% 左右因为系统提示词只需要传一次。第三个调优点是上下文窗口管理。Agent 执行过程中上下文会越来越长如果不加控制很快就会超出模型的最大上下文长度。我们的做法是实时监控上下文长度超过阈值时触发压缩。压缩策略有两种一是滑动窗口保留最近 N 条消息二是摘要压缩把旧消息用模型总结成一条摘要。滑动窗口简单但会丢失信息摘要压缩保留信息但增加一次模型调用。我们一般建议混合使用最近 10 条消息保留原文更早的消息压缩成摘要。class ContextManager: def __init__(self, max_tokens: int 8000, keep_recent: int 10): self.max_tokens max_tokens self.keep_recent keep_recent async def manage(self, messages: List[Message], provider: ModelProvider) - List[Message]: total_tokens sum(provider.count_tokens(m.content) for m in messages) if total_tokens self.max_tokens: return messages # 保留最近 N 条 recent messages[-self.keep_recent:] older messages[:-self.keep_recent] # 旧消息压缩成摘要 if older: summary await self._summarize(older, provider) return [Message(rolesystem, contentf历史摘要{summary})] recent return recent async def _summarize(self, messages: List[Message], provider: ModelProvider) - str: text \n.join(f{m.role}: {m.content} for m in messages) response await provider.chat([ Message(rolesystem, content请将以下对话历史压缩成简洁的摘要保留关键信息。), Message(roleuser, contenttext) ]) return response.content第四个调优点是工具执行异步化。工具执行往往是 IO 密集型的比如搜索、文件读写同步执行会阻塞整个 Agent 流程。我们把所有工具执行都改成异步的多个工具可以并行执行。这里有个细节并行执行工具的时候要注意工具之间的依赖关系有依赖的工具必须串行执行。我们的做法是在工具定义里加一个depends_on字段编排器根据依赖关系自动决定执行顺序。4.3 本地模型与云端模型的混合调度Orkas 支持同时接入云端模型和本地模型这是很多团队的实际需求。云端模型能力强但成本高、有延迟本地模型成本低、响应快但能力有限。混合调度的核心思路是简单任务用本地模型复杂任务用云端模型。实现上我们定义了一个HybridProvider内部维护一个路由策略。路由策略可以基于规则比如根据任务类型、输入长度、历史成功率也可以基于模型用一个小模型来判断任务复杂度。class HybridProvider(ModelProvider): def __init__(self, local: ModelProvider, cloud: ModelProvider, complexity_threshold: int 500): self.local local self.cloud cloud self.complexity_threshold complexity_threshold async def chat(self, messages: List[Message], **kwargs) - ChatResponse: # 简单策略根据输入长度判断 total_length sum(len(m.content) for m in messages) if total_length self.complexity_threshold: try: return await self.local.chat(messages, **kwargs) except Exception: # 本地模型失败降级到云端 return await self.cloud.chat(messages, **kwargs) else: return await self.cloud.chat(messages, **kwargs)这个策略看起来简单但实际用起来效果不错。我们的测试数据显示在客服问答场景下70% 的请求可以用本地模型处理只有 30% 的复杂请求需要云端模型整体成本降低了 60% 以上。这里有个坑要提醒本地模型和云端模型的输出格式可能不一致。本地模型尤其是小模型在工具调用格式上经常出错比如 JSON 格式不对、参数类型错误。我们的做法是在 Provider 层加一个输出校验和修复逻辑对本地模型的输出做格式校验不合法的话自动重试或者降级到云端模型。5. 常见问题与排查技巧实录5.1 工具调用串线问题问题现象多个 Agent 并发执行时Agent A 的工具调用结果返回给了 Agent B。排查思路首先检查工具注册表是否隔离。旧版全局注册表是常见原因新版每个 Agent 独立注册表一般不会出现这个问题。如果注册表已经隔离检查消息总线的路由逻辑确认receiver字段是否正确。最后检查工具执行的上下文传递确认correlation_id是否贯穿整个调用链。解决方案确保每个工具调用都携带correlation_id工具执行结果根据correlation_id路由回正确的 Agent。我们在ToolRegistry.execute方法里强制要求传入correlation_id执行结果通过消息总线发回。5.2 上下文窗口爆炸问题现象Agent 执行到一半突然报错“context length exceeded”。排查思路检查上下文管理器的配置确认max_tokens是否设置合理。检查是否有工具返回了超大结果比如搜索返回了完整网页内容。检查记忆检索是否返回了过多条目。解决方案设置合理的max_tokens一般建议模型最大上下文的 70%工具返回结果做截断处理比如搜索结果只保留摘要记忆检索限制top_k数量。另外可以在 Agent 的 system prompt 里明确要求模型精简输出。5.3 模型调用超时问题现象模型调用频繁超时尤其是高峰期。排查思路检查网络连接和 API 端点延迟。检查连接池配置是否合理。检查是否有大量并发请求同时发出。解决方案增加连接池大小配置合理的超时时间一般建议 30-60 秒实现请求队列和限流。对于超时请求配置自动重试和降级策略。5.4 常见问题速查表问题可能原因排查方法解决方案工具调用串线注册表未隔离/路由错误检查注册表和消息路由隔离注册表携带 correlation_id上下文爆炸未压缩/工具返回过大检查上下文长度和工具输出压缩上下文截断工具输出模型超时网络延迟/并发过高检查网络和并发量增加连接池配置重试降级状态污染共享内存/无隔离检查 Agent 状态存储Agent 级隔离消息传递通信记忆检索慢全量扫描/无索引检查记忆存储结构分层存储向量检索本地模型格式错误模型能力不足检查输出格式格式校验自动降级5.5 独家避坑技巧第一个技巧是给每个 Agent 加执行预算。Agent 执行过程中可能陷入死循环比如反复调用同一个工具设置一个最大执行步数或最大 token 消耗超过预算强制终止。我们的默认预算是 50 步或 100k token可以根据任务复杂度调整。第二个技巧是工具执行结果做缓存。同一个工具用相同参数调用多次结果应该是一样的幂等工具。我们加了一个工具结果缓存相同参数的调用直接返回缓存结果减少重复执行。这个优化在搜索类工具上效果特别明显能减少 40% 的重复调用。第三个技巧是Agent 之间通信加超时。Agent A 给 Agent B 发消息后如果 B 在超时时间内没有响应A 应该继续执行而不是无限等待。我们的做法是在消息总线上加超时控制超时后返回一个默认响应或者错误让调用方决定下一步。第四个技巧是定期做全链路压测。Agent 系统的性能瓶颈往往不在单个模块而在模块之间的交互。我们每个月做一次全链路压测模拟真实流量模式找出瓶颈点。压测的时候要注意模拟真实的工具调用延迟和模型响应延迟不然压测结果会偏乐观。6. 重构后的效果与后续演进方向重构完成后我们做了一轮完整的对比测试。在相同的硬件环境和任务集下新版 Orkas 的吞吐量提升了 3 倍平均响应延迟降低了 40%错误率从 5% 降到了 0.5% 以下。更重要的是代码的可维护性大幅提升新加一个 Agent 只需要实现接口和注册工具不用改核心逻辑。从架构层面看这次重构最大的价值是建立了清晰的扩展点。模型调用层可以接入任何模型供应商工具层可以注册任何工具编排层可以定义任何编排模式记忆层可以替换任何存储后端。这种插件化的设计让 Orkas 从一个“能用的框架”变成了一个“可演进的平台”。后续我们计划在几个方向继续演进。一是Agent 能力评估建立一套标准化的评估体系量化每个 Agent 的能力边界和可靠性。二是自适应编排根据任务特征自动选择最优的编排模式和 Agent 组合。三是安全增强在工具执行沙箱的基础上增加更细粒度的权限控制和审计日志。如果你也在做 Agent 框架的开发我的建议是不要等到问题堆积如山才重构在架构还清晰的时候就把分层和隔离做好。前期多花一周设计接口后期能省一个月修 bug。另外可观测性一定要从第一天就做没有 trace 的 Agent 系统就像没有日志的后端服务出了问题只能靠猜。