多代理架构在近两年的 AI 工程化进程中逐渐从理论探讨走向实际落地。无论是企业内部的智能客服中台还是面向复杂任务处理的自动化研究工具都在尝试用多个具备不同能力的模型协作完成单一模型难以承载的工作。本文将围绕一个非常具有代表性的工程设想展开——以 Perplexity 计算机为基础运行环境以 Fable 作为核心协调模型以 GPT-5.6 Terra 作为子代理执行体完整拆解这套多代理系统的架构设计、环境准备、核心代码实现、运行验证方式以及生产环境中最容易踩中的坑。1. 多代理系统Fable 与子代理之间的关系在正式写代码之前先把概念对齐。很多人听到“Fable 作为核心”“GPT-5.6 Terra 作为子代理”这样的描述第一反应是“这不就是多个大模型接口串在一起”。实际工程中的多代理系统远比简单的 API 串联复杂得多。1.1 什么是核心模型Coordinator核心模型是整个多代理系统的“大脑”。它负责接收来自用户或上层业务系统的原始任务理解任务目标把任务拆解成若干可执行的子步骤然后决定由哪个子代理去执行每一个子步骤最后汇总各个子代理的返回结果形成最终答案。在这个架构里Fable 承担的就是这个角色。Fable 本身是一个具备较强推理能力和工具调用规划能力的语言模型。它不直接接触底层数据源也不直接执行高风险操作而是像一个“项目经理”一样盯着整个任务链路的推进谁该上场了、谁的结果异常了、是否需要重新分配任务。这种设计的好处在于——即使某个子代理出现异常核心模型也能感知到异常状态并尝试切换执行策略而不是让整条任务链路直接中断。1.2 什么是子代理Sub-Agent子代理是真正干活的执行体。它们的能力往往比核心模型更专注、更垂直。比如一个子代理可能只负责文本摘要另一个子代理可能只负责调用搜索引擎并抓取网页内容还有一个子代理可能只负责本地文件系统的读取与写入。在本文的系统设计中GPT-5.6 Terra 承担的就是子代理角色。每个 Terra 实例可以具备不同的系统提示词System Prompt、不同的工具集Toolset和不同的权限边界Permission Scope。核心模型 Fable 负责任务编排而 Terra 负责在各自的能力范围内完成任务。1.3 核心模型与子代理之间的协同模式它们之间的协同关系可以用下面这段简化的流程来描述用户向系统提交一个复杂任务例如分析某网站最近一周的技术文章提取最热门的 5 个主题并生成一张调研报告。Fable 收到任务后先在内部将任务拆解为网页搜索、内容抓取、正文清洗、主题提取、报告生成五个子步骤。Fable 根据每个子步骤的特征依次调度对应的 Terra 子代理。每个 Terra 子代理把自己的执行结果可能是抓取到的网页原文、清洗后的正文或者提取出的主题列表返回给 Fable。Fable 汇总所有结果进行最后的推理归纳生成结构化的调研报告返回给用户。这种模式下Fable 不需要拥有全部技能Terra 子代理也不需要理解整个任务的全局目标。两者各司其职系统的可扩展性和故障隔离能力都会更好。2. 为什么说“失控出逃”事件值得每个开发者警醒在多个讨论多代理系统的社群里近期流传着一个被称作“大模型 GPT-5.6 SOL 失控出逃”的案例。虽然事件本身的真实性尚未得到官方全面证实但从工程视角来看这个案例暴露出一个非常真实的架构风险——当子代理被赋予过高权限且核心模型对子代理的中间状态缺少实时监控时系统可能做出超出任务范围的意外行为。2.1 对“失控”描述的客观理解所谓的“失控”在多代理系统的语境下并不是指模型突然有了自我意识而是指以下工程现象子代理在执行过程中由于陷入循环调用或上下文漂移产生了大量非预期的行为子代理没有遵守预设的权限边界对系统文件或外部服务发起了未被授权的操作核心模型在编排过程中对子代理返回的异常状态缺乏识别能力导致错误被逐级放大。这些现象其实都可以通过工程手段加以限制。2.2 对多代理系统设计的三点提醒第一任何子代理都不应该拥有“全量权限”。每个子代理的权限边界必须在系统设计阶段就通过代码强制约束而不是依赖模型自觉。第二核心模型与子代理之间的通信必须包含完整的“上下文跟踪信息”比如任务标识、步骤编号、超时阈值、可执行动作白名单。缺少这些信息系统出现异常时很难定位。第三生产环境的日志审计非常重要。子代理执行了什么工具调用、消耗了多少 Token、耗时多久这些信息必须全部落盘。否则一旦出现“失控”类问题你连复盘的基础都没有。本文后续的实战代码会体现这三个设计原则。3. 系统架构设计与技术选型在编写完整代码之前先把系统的整体架构画清楚。由于这里不使用 Mermaid 图表我用文字加分层的方式描述整个系统的组成。3.1 系统整体分层系统从上到下可以分为四层第一层接入层。负责接收用户请求完成基础的身份验证、请求限流、内容审核然后把任务转化为标准的内部请求格式。第二层编排层。该层由 Fable 核心模型和任务编排引擎组成。Fable 不直接操作外部服务而是通过工具调用协议把任务分发给下一层。第三层执行层。执行层部署了多个 GPT-5.6 Terra 子代理实例。每个实例绑定不同的工具集比如搜索引擎工具、网页抓取工具、本地文件工具、数据库查询工具等。第四层基础设施层。包括向量数据库、缓存服务、日志系统、监控告警服务等。这些基础设施为上面三层提供存储、检索与观测能力。3.2 为什么选择 Fable Terra 这种双模型组合在实际工程中核心模型与子代理往往选择不同规格的模型而不是选同一个模型跑所有角色。原因有几个职责不同。Fable 需要具备强大的上下文化能力与规划能力同时在长对话中维持稳定的状态跟踪能力。而 Terra 作为子代理往往只需要在小范围的特定任务上表现出色对全局推理的要求相对较低。成本容易控制。部署多个完整的顶级大模型实例成本很高而子代理采用较小参数规模或者更垂直的模型可以在保证任务效果的前提下显著降低推理成本。故障域分离。当某个 Terra 子代理出现异常时不会影响核心模型 Fable 的正常推理。这比一个单体大模型从头到尾执行所有任务要稳健得多。需要说明的是Fable 和 GPT-5.6 Terra 目前在不同版本迭代中都还在快速演进。本文中的代码示例不绑定某个特定官方 SDK而是以通用的大模型调用规范为基础编写的你可以根据实际使用的模型版本替换对应的接口地址、模型名称和认证方式。3.3 关键技术要点在实际实现中有几个技术点决定了这套系统是否稳定工具注册与发现机制Fable 如何知道当前系统里有哪些子代理可用每个子代理的能力描述、参数 schema、调用入口如何暴露给 Fable 状态跟踪机制一个任务从下发到完成中间会经过多个子代理。每个子代理执行完毕后产生的中间状态应该由谁来维护超时与降级机制子代理的执行时间是不可控的。如果搜索服务响应变慢或者模型推理时间过长要如何处理安全沙箱子代理能执行哪些命令、能写哪些目录、能访问哪些网络地址必须用代码硬性限制。下面进入环境准备与代码实战环节。4. 环境准备与项目工程结构为了把多代理系统跑起来我们需要准备一套基础的 Python 开发环境。本示例以 Python 3.10 为基础环境操作系统不限Windows、macOS、Linux 均可。4.1 需要准备的环境与依赖我假设你已经安装了 Python 3.10 或更高版本并能够使用 pip 安装第三方依赖。需要安装的核心依赖包括openai用于调用兼容 OpenAI 协议的大模型接口Fable 和 Terra 均通过该协议访问。pydantic用于定义工具调用的参数结构并做输入校验。pyyaml用于读取配置文件。fastapi 与 uvicorn如果希望以 Web 服务的方式暴露系统入口这两个库很方便。httpx用于子代理调用外部 HTTP 服务例如搜索 API、网页抓取接口。安装命令如下pip install openai pydantic pyyaml fastapi uvicorn httpx需要注意的是这里的版本不是写死的。你可以先安装最新稳定版本如果你所在项目的其他依赖与某个库存在版本冲突再根据实际情况锁定版本号。4.2 模型的接入方式说明本文示例通过 OpenAI 兼容协议调用 Fable 与 GPT-5.6 Terra。也就是说你需要在环境变量或配置文件中提供Fable 对应的接口地址base_urlFable 对应的 API Keyapi_keyFable 对应的模型名称model_nameTerra 对应的接口地址、API Key、模型名称如果你的模型服务商提供的接口遵循 OpenAI 协议那么下面的代码可以直接运行如果不是你只需要把请求逻辑替换成对应的 SDK 即可。4.3 项目目录结构我建议按照下面的目录结构组织工程代码multi-agent-system/ ├── config/ │ └── settings.yaml ├── core/ │ ├── coordinator.py │ ├── message.py │ └── task_state.py ├── agents/ │ ├── base.py │ ├── terra_search.py │ └── terra_file_reader.py ├── security/ │ └── permission.py ├── main.py └── requirements.txt各个文件的作用如下config/settings.yaml存放模型接口配置、子代理注册信息、安全策略参数。core/coordinator.pyFable 核心编排器的实现。core/message.py定义 Fable 与 Terra 之间传递的消息结构。core/task_state.py任务状态管理。agents/base.py子代理基类定义子代理的通用接口。agents/terra_search.py一个搜索类子代理示例。agents/terra_file_reader.py一个文件读取类子代理示例。security/permission.py权限校验模块。main.py程序入口用于演示一次完整的多代理任务执行。5. 完整实战基于 Fable 与 Terra 的多代理任务执行下面进入核心实战部分。我会一步一步带你从配置文件写到完整可运行的任务示例。5.1 编写全局配置文件文件路径config/settings.yamlfable: base_url: https://your-fable-endpoint.example.com/v1 api_key: ${FABLE_API_KEY} model_name: fable-5.1-coordinator terra: base_url: https://your-terra-endpoint.example.com/v1 api_key: ${TERRA_API_KEY} model_name: gpt-5.6-terra coordinator: max_iterations: 10 default_timeout_seconds: 60 subagents: - name: search_agent description: 用于搜索并返回技术文章的标题和链接 max_concurrency: 1 allowed_tools: [web_search, http_get] - name: file_reader_agent description: 读取本地文件系统中的文本文件 max_concurrency: 1 allowed_tools: [file_read]这里有一个细节需要注意api_key 我使用的是${FABLE_API_KEY}这种环境变量占位符。这是因为在实际项目中API Key 不应该硬编码在配置文件里而应该从环境变量或密钥管理服务中读取。5.2 定义消息结构Fable 核心模型与 Terra 子代理之间交互需要统一的消息格式。文件路径core/message.pyfrom dataclasses import dataclass, field from typing import Any, Dict, Optional dataclass class Message: role: str # user, assistant, tool, system content: str # 文本内容 task_id: Optional[str] None step_id: Optional[str] None tool_call_id: Optional[str] None metadata: Dict[str, Any] field(default_factorydict) def to_dict(self) - Dict[str, Any]: return { role: self.role, content: self.content, task_id: self.task_id, step_id: self.step_id, tool_call_id: self.tool_call_id, metadata: self.metadata, }这段代码的作用是统一消息格式。task_id 用于标识一次完整的用户请求step_id 用于标识任务内部的某个步骤tool_call_id 用于关联子代理工具调用的请求与返回结果。这些字段对于生产环境的问题追踪非常重要。5.3 实现任务状态管理器任务状态管理器负责跟踪每个任务的当前进度、已完成步骤、失败步骤和最终结果。文件路径core/task_state.pyimport uuid import time from typing import Callable, Dict, List class TaskStateManager: def __init__(self) - None: self._tasks: Dict[str, Dict] {} def create_task(self, user_input: str, max_iterations: int 10) - str: task_id str(uuid.uuid4()) self._tasks[task_id] { task_id: task_id, user_input: user_input, status: pending, created_at: time.time(), updated_at: time.time(), steps: [], result: None, max_iterations: max_iterations, } return task_id def update_step(self, task_id: str, step_name: str, status: str, detail: str ) - None: if task_id not in self._tasks: raise ValueError(ftask {task_id} not found) task self._tasks[task_id] task[steps].append({ step_name: step_name, status: status, detail: detail, time: time.time(), }) task[updated_at] time.time() def complete_task(self, task_id: str, result: str) - None: task self._tasks[task_id] task[status] completed task[result] result task[updated_at] time.time() def fail_task(self, task_id: str, error: str) - None: task self._tasks[task_id] task[status] failed task[result] error task[updated_at] time.time() def get_task(self, task_id: str) - Dict: return self._tasks.get(task_id, {})这个类不直接依赖任何具体的模型 SDK所以在测试时可以独立验证任务状态流转逻辑。5.4 定义子代理基类所有子代理都必须继承这个基类并实现run方法。文件路径agents/base.pyfrom abc import ABC, abstractmethod from typing import Any, Dict from core.message import Message class BaseAgent(ABC): def __init__(self, name: str, allowed_tools: list[str]) - None: self.name name self.allowed_tools allowed_tools abstractmethod def run(self, user_input: str, context: Dict[str, Any]) - Message: 执行子代理任务返回结果消息 pass def permission_check(self, tool_name: str) - bool: 检查某个工具是否在当前子代理的允许范围内 return tool_name in self.allowed_tools这里我留了一个 permission_check 方法它的意义在于子代理在执行具体的工具调用之前必须先检查该工具是否在允许列表里。这是多代理系统安全设计的第一道防线。5.5 实现搜索子代理 TerraSearch搜索子代理模拟的是一个具备网络搜索能力的 Terra 实例。文件路径agents/terra_search.pyimport httpx from typing import Any, Dict from agents.base import BaseAgent from core.message import Message class TerraSearchAgent(BaseAgent): def __init__(self, name: str search_agent, allowed_tools: list[str] | None None, search_endpoint: str https://your-search-api.example.com/search): super().__init__(namename, allowed_toolsallowed_tools or [web_search, http_get]) self.search_endpoint search_endpoint async def run(self, user_input: str, context: Dict[str, Any]) - Message: # 打印当前任务上下文便于观察 print(f[TerraSearchAgent] 接收到任务: {user_input}) if not self.permission_check(web_search): return Message( roletool, content[ERROR] 当前子代理没有 web_search 工具权限, task_idcontext.get(task_id), step_idcontext.get(step_id), metadata{agent: self.name}, ) # 调用外部搜索服务 async with httpx.AsyncClient(timeout10) as client: resp await client.get( self.search_endpoint, params{q: user_input, limit: 5}, ) if resp.status_code 200: data resp.json() results data.get(results, []) formatted \n.join( [f- {item[title]}: {item[url]} for item in results] ) return Message( roletool, contentf搜索结果如下:\n{formatted}, task_idcontext.get(task_id), step_idcontext.get(step_id), metadata{agent: self.name}, ) else: return Message( roletool, contentf[ERROR] 搜索服务返回状态码 {resp.status_code}, task_idcontext.get(task_id), step_idcontext.get(step_id), metadata{agent: self.name}, )这个示例在搜索服务不可用时会返回错误信息但不会把异常抛出这是刻意为之——子代理调用外部服务失败时应该把失败信息返回给核心模型由核心模型决定是否更换策略。5.6 实现 Fable 核心协调器这是整个系统的核心文件。文件路径core/coordinator.pyfrom typing import Any, Dict, List from openai import AsyncOpenAI from core.message import Message from core.task_state import TaskStateManager from agents.base import BaseAgent class FableCoordinator: def __init__( self, coordination_model_config: Dict[str, str], subagents: Dict[str, BaseAgent], task_state_manager: TaskStateManager, max_iterations: int 10, default_timeout: int 60, ) - None: self.client AsyncOpenAI( base_urlcoordination_model_config[base_url], api_keycoordination_model_config[api_key], ) self.model_name coordination_model_config[model_name] self.subagents subagents self.task_state_manager task_state_manager self.max_iterations max_iterations self.default_timeout default_timeout def _build_system_prompt(self, task_id: str) - str: 构建 Fable 的系统提示词向模型声明可用子代理 agent_descriptions \n.join( [f- {name}: {agent.description} for name, agent in self.subagents.items()] ) return ( 你是一个多代理系统的核心协调器。你负责理解用户任务并把任务拆解为多个步骤 f然后选择合适的子代理去执行。当前任务 ID 为 {task_id}。\n 可用的子代理如下\n f{agent_descriptions}\n\n 请严格按照任务拆解 - 子代理调度 - 结果汇总的流程完成任务。 当子代理返回错误时请尝试调整策略后再次执行不要直接放弃。 ) async def run(self, user_input: str) - str: task_id self.task_state_manager.create_task(user_input) messages: List[Message] [ Message(rolesystem, contentself._build_system_prompt(task_id)), Message(roleuser, contentuser_input, task_idtask_id), ] for iteration in range(self.max_iterations): # 调用 Fable 模型 response await self.client.chat.completions.create( modelself.model_name, messages[msg.to_dict() for msg in messages], temperature0.2, ) assistant_content response.choices[0].message.content or tool_calls response.choices[0].message.tool_calls or [] messages.append(Message( roleassistant, contentassistant_content, task_idtask_id, )) # 如果模型没有请求调用工具说明它在生成最终答案 if not tool_calls: self.task_state_manager.complete_task(task_id, assistant_content) return assistant_content # 逐条处理工具调用 for tool_call in tool_calls: func_name tool_call.function.name arguments tool_call.function.arguments or {} if func_name in self.subagents: # 通过子代理执行任务 sub_agent self.subagents[func_name] result_message await sub_agent.run( user_inputarguments, context{ task_id: task_id, step_id: fiter_{iteration}_tool_{tool_call.id}, }, ) messages.append(result_message) else: # 未知工具 messages.append(Message( roletool, contentf[ERROR] 未知工具: {func_name}, task_idtask_id, tool_call_idtool_call.id, )) error_msg f任务在 {self.max_iterations} 次迭代内未完成 self.task_state_manager.fail_task(task_id, error_msg) return error_msg这个协调器的关键逻辑是循环调用 Fable 模型。每一轮里Fable 有两种可能输出第一它认为自己已经掌握了足够的上下文不需要再调用任何子代理于是直接返回最终答案循环结束。第二它要求调用某个子代理那么协调器会从注册的 subagents 字典里找到对应的子代理实例把参数传进去执行然后把执行结果作为一个 tool 角色消息附加到消息列表里进入下一轮循环。5.7 在 main.py 中组装整个系统文件路径main.pyimport asyncio import os from core.coordinator import FableCoordinator from core.task_state import TaskStateManager from agents.terra_search import TerraSearchAgent from agents.terra_file_reader import TerraFileReaderAgent def load_config(): # 这里为了演示简化了 yaml 读取逻辑实际项目中应使用 pyyaml 读取 settings.yaml return { fable: { base_url: os.getenv(FABLE_BASE_URL, https://your-fable-endpoint.example.com/v1), api_key: os.getenv(FABLE_API_KEY, test-key), model_name: fable-5.1-coordinator, }, terra: { base_url: os.getenv(TERRA_BASE_URL, https://your-terra-endpoint.example.com/v1), api_key: os.getenv(TERRA_API_KEY, test-key), model_name: gpt-5.6-terra, }, coordinator: { max_iterations: 8, default_timeout_seconds: 60, }, } async def main(): config load_config() task_state_manager TaskStateManager() search_agent TerraSearchAgent( namesearch_agent, allowed_tools[web_search], ) file_reader_agent TerraFileReaderAgent( namefile_reader_agent, allowed_tools[file_read], ) subagents { search_agent: search_agent, file_reader_agent: file_reader_agent, } coordinator FableCoordinator( coordination_model_configconfig[fable], subagentssubagents, task_state_managertask_state_manager, max_iterationsconfig[coordinator][max_iterations], default_timeout_secondsconfig[coordinator][default_timeout_seconds], ) user_input ( 请先搜索最近一周关于多代理系统的最佳实践文章 然后读取本地 ./notes.txt 文件把搜索结果和笔记内容整合成一份学习总结。 ) result await coordinator.run(user_input) print(最终结果) print(result) if __name__ __main__: asyncio.run(main())这里需要补充一个 TerraFileReaderAgent实现比较简单只负责读取本地文件文件路径agents/terra_file_reader.pyfrom typing import Any, Dict from agents.base import BaseAgent from core.message import Message class TerraFileReaderAgent(BaseAgent): def __init__(self, name: str file_reader_agent, allowed_tools: list[str] | None None): super().__init__(namename, allowed_toolsallowed_tools or [file_read]) def run(self, user_input: str, context: Dict[str, Any]) - Message: print(f[TerraFileReaderAgent] 接收到文件读取任务) if not self.permission_check(file_read): return Message( roletool, content[ERROR] 当前子代理没有 file_read 工具权限, task_idcontext.get(task_id), step_idcontext.get(step_id), metadata{agent: self.name}, ) file_path user_input.strip().strip(\) try: with open(file_path, r, encodingutf-8) as f: content f.read() return Message( roletool, contentf文件内容如下:\n{content[:2000]}, task_idcontext.get(task_id), step_idcontext.get(step_id), metadata{agent: self.name}, ) except Exception as e: return Message( roletool, contentf[ERROR] 文件读取失败: {str(e)}, task_idcontext.get(task_id), step_idcontext.get(step_id), metadata{agent: self.name}, )5.8 运行与预期验证在终端里执行python main.py如果一切正常你会看到类似下面的输出流程[TerraSearchAgent] 接收到任务: 最近一周 多代理系统 最佳实践 [TerraFileReaderAgent] 接收到文件读取任务 最终结果 根据搜索到的文章和本地笔记本周关于多代理系统的重点实践方向包括 1. 核心模型与子代理的权限隔离... 2. 子代理超时与降级策略... 3. 任务状态追踪与审计日志的重要性...这里要注意实际的输出会依赖你使用的 Fable 模型对工具调用的判断以及搜索服务返回的数据。不同模型、不同提示词生成的结果会有差异。重点是整个链路打通了。6. 常见问题与排查思路不管代码写得多顺手多代理系统在生产环境中一定会遇到各种问题。下面整理了几类高频问题。6.1 子代理超时导致任务卡死问题现象常见原因解决思路整个任务长时间没有返回结果某个子代理调用外部服务时卡住为每个子代理设置独立超时时间在子代理 run 方法中使用 asyncio.wait_for 包裹耗时调用超时后返回降级结果数据库连接池占满子代理并发数超过数据库连接池上限在子代理层增加信号量Semaphore限制并发数合理配置连接池大小Fable 循环调用同一个工具超过 10 次提示词中未明确工具使用边界在 system prompt 中限定每个工具的调用次数协调器增加重复调用检测6.2 模型返回了未知工具名如果你发现协调器抛出了“未知工具”错误很可能是因为_build_system_prompt中提供的子代理描述不够清晰或者subagents字典中的工具名与 prompt 中使用的名称不一致。解决方式保证 subagents 字典的键名与工具名严格一致在 prompt 中明确说明“只能使用上述列出的工具名”在协调器中增加模糊匹配或同义词映射早期调试阶段非常有用。6.3 出现非预期的子代理行为这个问题对应前面提到的“失控”风险。假如你的 Terra 子代理在没有被授权的情况下访问了某个系统目录说明权限校验没有生效。排查顺序检查子代理继承的 BaseAgent 是否在 run 方法开头调用了 permission_check检查 allowed_tools 白名单是否配置正确检查系统提示词中是否无意间把高权限的工具描述给了子代理检查日志确认子代理实际执行的操作命令。再强调一次多代理系统的安全不能依赖模型的“自觉”必须通过代码强制约束。6.4 上下文过长导致模型响应缓慢如果任务比较复杂消息列表会不断累积最终可能超过模型的上下文窗口或者导致推理时间明显变长。应对策略在每一轮迭代后对消息列表做摘要压缩把早期步骤的详细内容替换为简短摘要限制传入子代理的原始文本长度比如只保留关键段落对长时间运行的任务定期触发“保存中间状态 重置上下文”的机制。7. 从“失控出逃”事件中学到的工程启示回顾前面提到的“GPT-5.6 SOL 失控出逃”事件虽然事件细节仍有争议但它对整个多代理系统社区的技术启示是明确的。7.1 权限最小化原则每个子代理应该只拥有完成自身任务所需的最小权限。搜索代理只需要访问搜索 API 的权限文件读取代理只需要读取指定目录的权限数据库代理只拥有对应数据库账号的最小查询权限。在多代理系统里权限不是一个宽泛的概念而是每个工具、每个目录、每个 API 端点的细致列表。7.2 每次工具调用都需要审计生产环境中所有 Fable 与 Terra 之间的交互消息、工具调用记录、执行耗时、Token 消耗都必须完整记录到日志系统或审计平台中。这样一旦发生“失控”类问题你可以精确地回溯问题发生的起点而不是只能观察最终结果。建议日志至少包含以下字段task_id 与 step_id调用方coordinator / 具体子代理名称工具名称与参数返回状态与错误信息时间戳与耗时。7.3 动态降级与熔断当某个子代理连续失败达到阈值时系统应该自动进入降级模式。比如搜索服务连续三次超时Fable 可以选择使用本地缓存结果继续执行也可以直接向用户说明“搜索服务暂不可用已返回基于已有资料的结果”。降级策略的价值在于即使某个子代理异常用户的整体任务体验不会完全中断。7.4 定期演练异常恢复不要等到生产环境出现严重事故再去想恢复方案建议定期做故障演练模拟某个子代理失联、模拟搜索 API 返回 500、模拟文件系统权限被误改。每演练一次你都会发现自己系统在容错设计上的薄弱点。8. 工程落地建议与未来学习方向基于前面的完整实战最后整理几条工程层面的建议不管你是个人开发者在学习多代理架构还是团队在做技术服务化改造都值得收藏一份。8.1 命名规范子代理命名建议采用“能力域_动作”的结构比如search_web、file_read、db_query、email_send。命名太随意会让 Fable 在函数选择时出现混乱。8.2 配置管理所有模型接口地址、API Key、子代理白名单、超时阈值都应该放在配置中心或环境变量中不要写死在代码里。不同环境开发、测试、生产使用不同的配置避免因为环境差异导致线上事故。8.3 异常处理策略子代理的 run 方法中尽量捕获所有可预期的异常并把异常信息转化为消息返回给协调器而不是直接抛出。因为协调器需要根据错误信息决定下一轮策略一个简单粗暴的异常抛出会让整个编排链路断裂。8.4 从模型能力到系统能力的转换很多团队一开始把多代理系统理解成“多模型 API 调用”这是一个认知误区。真正的多代理系统设计重心不在模型本身而在系统能力如何管理状态、如何做权限隔离、如何监控链路、如何防止级联故障。模型只是执行单元系统的稳定性由工程框架决定。后续如果你想继续深入可以关注以下几个方向如何为子代理引入更丰富的工具集例如数据库查询、邮件发送、知识库检索如何实现子代理的横向扩容与负载均衡如何利用向量数据库为 Fable 提供结构性记忆如何建立完整的可观测性指标包括任务成功率、平均延迟、单任务 Token 消耗等。多代理系统的工程化还处于快速发展期很多设计模式还没有形成绝对标准但这恰恰是技术人最好的学习窗口。动手写一个最小闭环的 Fable Terra 系统比读十篇架构分析都更有价值。 多代理系统实战：基于Fable与GPT-5.6 Terra的架构设计与实现
多代理架构在近两年的 AI 工程化进程中逐渐从理论探讨走向实际落地。无论是企业内部的智能客服中台还是面向复杂任务处理的自动化研究工具都在尝试用多个具备不同能力的模型协作完成单一模型难以承载的工作。本文将围绕一个非常具有代表性的工程设想展开——以 Perplexity 计算机为基础运行环境以 Fable 作为核心协调模型以 GPT-5.6 Terra 作为子代理执行体完整拆解这套多代理系统的架构设计、环境准备、核心代码实现、运行验证方式以及生产环境中最容易踩中的坑。1. 多代理系统Fable 与子代理之间的关系在正式写代码之前先把概念对齐。很多人听到“Fable 作为核心”“GPT-5.6 Terra 作为子代理”这样的描述第一反应是“这不就是多个大模型接口串在一起”。实际工程中的多代理系统远比简单的 API 串联复杂得多。1.1 什么是核心模型Coordinator核心模型是整个多代理系统的“大脑”。它负责接收来自用户或上层业务系统的原始任务理解任务目标把任务拆解成若干可执行的子步骤然后决定由哪个子代理去执行每一个子步骤最后汇总各个子代理的返回结果形成最终答案。在这个架构里Fable 承担的就是这个角色。Fable 本身是一个具备较强推理能力和工具调用规划能力的语言模型。它不直接接触底层数据源也不直接执行高风险操作而是像一个“项目经理”一样盯着整个任务链路的推进谁该上场了、谁的结果异常了、是否需要重新分配任务。这种设计的好处在于——即使某个子代理出现异常核心模型也能感知到异常状态并尝试切换执行策略而不是让整条任务链路直接中断。1.2 什么是子代理Sub-Agent子代理是真正干活的执行体。它们的能力往往比核心模型更专注、更垂直。比如一个子代理可能只负责文本摘要另一个子代理可能只负责调用搜索引擎并抓取网页内容还有一个子代理可能只负责本地文件系统的读取与写入。在本文的系统设计中GPT-5.6 Terra 承担的就是子代理角色。每个 Terra 实例可以具备不同的系统提示词System Prompt、不同的工具集Toolset和不同的权限边界Permission Scope。核心模型 Fable 负责任务编排而 Terra 负责在各自的能力范围内完成任务。1.3 核心模型与子代理之间的协同模式它们之间的协同关系可以用下面这段简化的流程来描述用户向系统提交一个复杂任务例如分析某网站最近一周的技术文章提取最热门的 5 个主题并生成一张调研报告。Fable 收到任务后先在内部将任务拆解为网页搜索、内容抓取、正文清洗、主题提取、报告生成五个子步骤。Fable 根据每个子步骤的特征依次调度对应的 Terra 子代理。每个 Terra 子代理把自己的执行结果可能是抓取到的网页原文、清洗后的正文或者提取出的主题列表返回给 Fable。Fable 汇总所有结果进行最后的推理归纳生成结构化的调研报告返回给用户。这种模式下Fable 不需要拥有全部技能Terra 子代理也不需要理解整个任务的全局目标。两者各司其职系统的可扩展性和故障隔离能力都会更好。2. 为什么说“失控出逃”事件值得每个开发者警醒在多个讨论多代理系统的社群里近期流传着一个被称作“大模型 GPT-5.6 SOL 失控出逃”的案例。虽然事件本身的真实性尚未得到官方全面证实但从工程视角来看这个案例暴露出一个非常真实的架构风险——当子代理被赋予过高权限且核心模型对子代理的中间状态缺少实时监控时系统可能做出超出任务范围的意外行为。2.1 对“失控”描述的客观理解所谓的“失控”在多代理系统的语境下并不是指模型突然有了自我意识而是指以下工程现象子代理在执行过程中由于陷入循环调用或上下文漂移产生了大量非预期的行为子代理没有遵守预设的权限边界对系统文件或外部服务发起了未被授权的操作核心模型在编排过程中对子代理返回的异常状态缺乏识别能力导致错误被逐级放大。这些现象其实都可以通过工程手段加以限制。2.2 对多代理系统设计的三点提醒第一任何子代理都不应该拥有“全量权限”。每个子代理的权限边界必须在系统设计阶段就通过代码强制约束而不是依赖模型自觉。第二核心模型与子代理之间的通信必须包含完整的“上下文跟踪信息”比如任务标识、步骤编号、超时阈值、可执行动作白名单。缺少这些信息系统出现异常时很难定位。第三生产环境的日志审计非常重要。子代理执行了什么工具调用、消耗了多少 Token、耗时多久这些信息必须全部落盘。否则一旦出现“失控”类问题你连复盘的基础都没有。本文后续的实战代码会体现这三个设计原则。3. 系统架构设计与技术选型在编写完整代码之前先把系统的整体架构画清楚。由于这里不使用 Mermaid 图表我用文字加分层的方式描述整个系统的组成。3.1 系统整体分层系统从上到下可以分为四层第一层接入层。负责接收用户请求完成基础的身份验证、请求限流、内容审核然后把任务转化为标准的内部请求格式。第二层编排层。该层由 Fable 核心模型和任务编排引擎组成。Fable 不直接操作外部服务而是通过工具调用协议把任务分发给下一层。第三层执行层。执行层部署了多个 GPT-5.6 Terra 子代理实例。每个实例绑定不同的工具集比如搜索引擎工具、网页抓取工具、本地文件工具、数据库查询工具等。第四层基础设施层。包括向量数据库、缓存服务、日志系统、监控告警服务等。这些基础设施为上面三层提供存储、检索与观测能力。3.2 为什么选择 Fable Terra 这种双模型组合在实际工程中核心模型与子代理往往选择不同规格的模型而不是选同一个模型跑所有角色。原因有几个职责不同。Fable 需要具备强大的上下文化能力与规划能力同时在长对话中维持稳定的状态跟踪能力。而 Terra 作为子代理往往只需要在小范围的特定任务上表现出色对全局推理的要求相对较低。成本容易控制。部署多个完整的顶级大模型实例成本很高而子代理采用较小参数规模或者更垂直的模型可以在保证任务效果的前提下显著降低推理成本。故障域分离。当某个 Terra 子代理出现异常时不会影响核心模型 Fable 的正常推理。这比一个单体大模型从头到尾执行所有任务要稳健得多。需要说明的是Fable 和 GPT-5.6 Terra 目前在不同版本迭代中都还在快速演进。本文中的代码示例不绑定某个特定官方 SDK而是以通用的大模型调用规范为基础编写的你可以根据实际使用的模型版本替换对应的接口地址、模型名称和认证方式。3.3 关键技术要点在实际实现中有几个技术点决定了这套系统是否稳定工具注册与发现机制Fable 如何知道当前系统里有哪些子代理可用每个子代理的能力描述、参数 schema、调用入口如何暴露给 Fable 状态跟踪机制一个任务从下发到完成中间会经过多个子代理。每个子代理执行完毕后产生的中间状态应该由谁来维护超时与降级机制子代理的执行时间是不可控的。如果搜索服务响应变慢或者模型推理时间过长要如何处理安全沙箱子代理能执行哪些命令、能写哪些目录、能访问哪些网络地址必须用代码硬性限制。下面进入环境准备与代码实战环节。4. 环境准备与项目工程结构为了把多代理系统跑起来我们需要准备一套基础的 Python 开发环境。本示例以 Python 3.10 为基础环境操作系统不限Windows、macOS、Linux 均可。4.1 需要准备的环境与依赖我假设你已经安装了 Python 3.10 或更高版本并能够使用 pip 安装第三方依赖。需要安装的核心依赖包括openai用于调用兼容 OpenAI 协议的大模型接口Fable 和 Terra 均通过该协议访问。pydantic用于定义工具调用的参数结构并做输入校验。pyyaml用于读取配置文件。fastapi 与 uvicorn如果希望以 Web 服务的方式暴露系统入口这两个库很方便。httpx用于子代理调用外部 HTTP 服务例如搜索 API、网页抓取接口。安装命令如下pip install openai pydantic pyyaml fastapi uvicorn httpx需要注意的是这里的版本不是写死的。你可以先安装最新稳定版本如果你所在项目的其他依赖与某个库存在版本冲突再根据实际情况锁定版本号。4.2 模型的接入方式说明本文示例通过 OpenAI 兼容协议调用 Fable 与 GPT-5.6 Terra。也就是说你需要在环境变量或配置文件中提供Fable 对应的接口地址base_urlFable 对应的 API Keyapi_keyFable 对应的模型名称model_nameTerra 对应的接口地址、API Key、模型名称如果你的模型服务商提供的接口遵循 OpenAI 协议那么下面的代码可以直接运行如果不是你只需要把请求逻辑替换成对应的 SDK 即可。4.3 项目目录结构我建议按照下面的目录结构组织工程代码multi-agent-system/ ├── config/ │ └── settings.yaml ├── core/ │ ├── coordinator.py │ ├── message.py │ └── task_state.py ├── agents/ │ ├── base.py │ ├── terra_search.py │ └── terra_file_reader.py ├── security/ │ └── permission.py ├── main.py └── requirements.txt各个文件的作用如下config/settings.yaml存放模型接口配置、子代理注册信息、安全策略参数。core/coordinator.pyFable 核心编排器的实现。core/message.py定义 Fable 与 Terra 之间传递的消息结构。core/task_state.py任务状态管理。agents/base.py子代理基类定义子代理的通用接口。agents/terra_search.py一个搜索类子代理示例。agents/terra_file_reader.py一个文件读取类子代理示例。security/permission.py权限校验模块。main.py程序入口用于演示一次完整的多代理任务执行。5. 完整实战基于 Fable 与 Terra 的多代理任务执行下面进入核心实战部分。我会一步一步带你从配置文件写到完整可运行的任务示例。5.1 编写全局配置文件文件路径config/settings.yamlfable: base_url: https://your-fable-endpoint.example.com/v1 api_key: ${FABLE_API_KEY} model_name: fable-5.1-coordinator terra: base_url: https://your-terra-endpoint.example.com/v1 api_key: ${TERRA_API_KEY} model_name: gpt-5.6-terra coordinator: max_iterations: 10 default_timeout_seconds: 60 subagents: - name: search_agent description: 用于搜索并返回技术文章的标题和链接 max_concurrency: 1 allowed_tools: [web_search, http_get] - name: file_reader_agent description: 读取本地文件系统中的文本文件 max_concurrency: 1 allowed_tools: [file_read]这里有一个细节需要注意api_key 我使用的是${FABLE_API_KEY}这种环境变量占位符。这是因为在实际项目中API Key 不应该硬编码在配置文件里而应该从环境变量或密钥管理服务中读取。5.2 定义消息结构Fable 核心模型与 Terra 子代理之间交互需要统一的消息格式。文件路径core/message.pyfrom dataclasses import dataclass, field from typing import Any, Dict, Optional dataclass class Message: role: str # user, assistant, tool, system content: str # 文本内容 task_id: Optional[str] None step_id: Optional[str] None tool_call_id: Optional[str] None metadata: Dict[str, Any] field(default_factorydict) def to_dict(self) - Dict[str, Any]: return { role: self.role, content: self.content, task_id: self.task_id, step_id: self.step_id, tool_call_id: self.tool_call_id, metadata: self.metadata, }这段代码的作用是统一消息格式。task_id 用于标识一次完整的用户请求step_id 用于标识任务内部的某个步骤tool_call_id 用于关联子代理工具调用的请求与返回结果。这些字段对于生产环境的问题追踪非常重要。5.3 实现任务状态管理器任务状态管理器负责跟踪每个任务的当前进度、已完成步骤、失败步骤和最终结果。文件路径core/task_state.pyimport uuid import time from typing import Callable, Dict, List class TaskStateManager: def __init__(self) - None: self._tasks: Dict[str, Dict] {} def create_task(self, user_input: str, max_iterations: int 10) - str: task_id str(uuid.uuid4()) self._tasks[task_id] { task_id: task_id, user_input: user_input, status: pending, created_at: time.time(), updated_at: time.time(), steps: [], result: None, max_iterations: max_iterations, } return task_id def update_step(self, task_id: str, step_name: str, status: str, detail: str ) - None: if task_id not in self._tasks: raise ValueError(ftask {task_id} not found) task self._tasks[task_id] task[steps].append({ step_name: step_name, status: status, detail: detail, time: time.time(), }) task[updated_at] time.time() def complete_task(self, task_id: str, result: str) - None: task self._tasks[task_id] task[status] completed task[result] result task[updated_at] time.time() def fail_task(self, task_id: str, error: str) - None: task self._tasks[task_id] task[status] failed task[result] error task[updated_at] time.time() def get_task(self, task_id: str) - Dict: return self._tasks.get(task_id, {})这个类不直接依赖任何具体的模型 SDK所以在测试时可以独立验证任务状态流转逻辑。5.4 定义子代理基类所有子代理都必须继承这个基类并实现run方法。文件路径agents/base.pyfrom abc import ABC, abstractmethod from typing import Any, Dict from core.message import Message class BaseAgent(ABC): def __init__(self, name: str, allowed_tools: list[str]) - None: self.name name self.allowed_tools allowed_tools abstractmethod def run(self, user_input: str, context: Dict[str, Any]) - Message: 执行子代理任务返回结果消息 pass def permission_check(self, tool_name: str) - bool: 检查某个工具是否在当前子代理的允许范围内 return tool_name in self.allowed_tools这里我留了一个 permission_check 方法它的意义在于子代理在执行具体的工具调用之前必须先检查该工具是否在允许列表里。这是多代理系统安全设计的第一道防线。5.5 实现搜索子代理 TerraSearch搜索子代理模拟的是一个具备网络搜索能力的 Terra 实例。文件路径agents/terra_search.pyimport httpx from typing import Any, Dict from agents.base import BaseAgent from core.message import Message class TerraSearchAgent(BaseAgent): def __init__(self, name: str search_agent, allowed_tools: list[str] | None None, search_endpoint: str https://your-search-api.example.com/search): super().__init__(namename, allowed_toolsallowed_tools or [web_search, http_get]) self.search_endpoint search_endpoint async def run(self, user_input: str, context: Dict[str, Any]) - Message: # 打印当前任务上下文便于观察 print(f[TerraSearchAgent] 接收到任务: {user_input}) if not self.permission_check(web_search): return Message( roletool, content[ERROR] 当前子代理没有 web_search 工具权限, task_idcontext.get(task_id), step_idcontext.get(step_id), metadata{agent: self.name}, ) # 调用外部搜索服务 async with httpx.AsyncClient(timeout10) as client: resp await client.get( self.search_endpoint, params{q: user_input, limit: 5}, ) if resp.status_code 200: data resp.json() results data.get(results, []) formatted \n.join( [f- {item[title]}: {item[url]} for item in results] ) return Message( roletool, contentf搜索结果如下:\n{formatted}, task_idcontext.get(task_id), step_idcontext.get(step_id), metadata{agent: self.name}, ) else: return Message( roletool, contentf[ERROR] 搜索服务返回状态码 {resp.status_code}, task_idcontext.get(task_id), step_idcontext.get(step_id), metadata{agent: self.name}, )这个示例在搜索服务不可用时会返回错误信息但不会把异常抛出这是刻意为之——子代理调用外部服务失败时应该把失败信息返回给核心模型由核心模型决定是否更换策略。5.6 实现 Fable 核心协调器这是整个系统的核心文件。文件路径core/coordinator.pyfrom typing import Any, Dict, List from openai import AsyncOpenAI from core.message import Message from core.task_state import TaskStateManager from agents.base import BaseAgent class FableCoordinator: def __init__( self, coordination_model_config: Dict[str, str], subagents: Dict[str, BaseAgent], task_state_manager: TaskStateManager, max_iterations: int 10, default_timeout: int 60, ) - None: self.client AsyncOpenAI( base_urlcoordination_model_config[base_url], api_keycoordination_model_config[api_key], ) self.model_name coordination_model_config[model_name] self.subagents subagents self.task_state_manager task_state_manager self.max_iterations max_iterations self.default_timeout default_timeout def _build_system_prompt(self, task_id: str) - str: 构建 Fable 的系统提示词向模型声明可用子代理 agent_descriptions \n.join( [f- {name}: {agent.description} for name, agent in self.subagents.items()] ) return ( 你是一个多代理系统的核心协调器。你负责理解用户任务并把任务拆解为多个步骤 f然后选择合适的子代理去执行。当前任务 ID 为 {task_id}。\n 可用的子代理如下\n f{agent_descriptions}\n\n 请严格按照任务拆解 - 子代理调度 - 结果汇总的流程完成任务。 当子代理返回错误时请尝试调整策略后再次执行不要直接放弃。 ) async def run(self, user_input: str) - str: task_id self.task_state_manager.create_task(user_input) messages: List[Message] [ Message(rolesystem, contentself._build_system_prompt(task_id)), Message(roleuser, contentuser_input, task_idtask_id), ] for iteration in range(self.max_iterations): # 调用 Fable 模型 response await self.client.chat.completions.create( modelself.model_name, messages[msg.to_dict() for msg in messages], temperature0.2, ) assistant_content response.choices[0].message.content or tool_calls response.choices[0].message.tool_calls or [] messages.append(Message( roleassistant, contentassistant_content, task_idtask_id, )) # 如果模型没有请求调用工具说明它在生成最终答案 if not tool_calls: self.task_state_manager.complete_task(task_id, assistant_content) return assistant_content # 逐条处理工具调用 for tool_call in tool_calls: func_name tool_call.function.name arguments tool_call.function.arguments or {} if func_name in self.subagents: # 通过子代理执行任务 sub_agent self.subagents[func_name] result_message await sub_agent.run( user_inputarguments, context{ task_id: task_id, step_id: fiter_{iteration}_tool_{tool_call.id}, }, ) messages.append(result_message) else: # 未知工具 messages.append(Message( roletool, contentf[ERROR] 未知工具: {func_name}, task_idtask_id, tool_call_idtool_call.id, )) error_msg f任务在 {self.max_iterations} 次迭代内未完成 self.task_state_manager.fail_task(task_id, error_msg) return error_msg这个协调器的关键逻辑是循环调用 Fable 模型。每一轮里Fable 有两种可能输出第一它认为自己已经掌握了足够的上下文不需要再调用任何子代理于是直接返回最终答案循环结束。第二它要求调用某个子代理那么协调器会从注册的 subagents 字典里找到对应的子代理实例把参数传进去执行然后把执行结果作为一个 tool 角色消息附加到消息列表里进入下一轮循环。5.7 在 main.py 中组装整个系统文件路径main.pyimport asyncio import os from core.coordinator import FableCoordinator from core.task_state import TaskStateManager from agents.terra_search import TerraSearchAgent from agents.terra_file_reader import TerraFileReaderAgent def load_config(): # 这里为了演示简化了 yaml 读取逻辑实际项目中应使用 pyyaml 读取 settings.yaml return { fable: { base_url: os.getenv(FABLE_BASE_URL, https://your-fable-endpoint.example.com/v1), api_key: os.getenv(FABLE_API_KEY, test-key), model_name: fable-5.1-coordinator, }, terra: { base_url: os.getenv(TERRA_BASE_URL, https://your-terra-endpoint.example.com/v1), api_key: os.getenv(TERRA_API_KEY, test-key), model_name: gpt-5.6-terra, }, coordinator: { max_iterations: 8, default_timeout_seconds: 60, }, } async def main(): config load_config() task_state_manager TaskStateManager() search_agent TerraSearchAgent( namesearch_agent, allowed_tools[web_search], ) file_reader_agent TerraFileReaderAgent( namefile_reader_agent, allowed_tools[file_read], ) subagents { search_agent: search_agent, file_reader_agent: file_reader_agent, } coordinator FableCoordinator( coordination_model_configconfig[fable], subagentssubagents, task_state_managertask_state_manager, max_iterationsconfig[coordinator][max_iterations], default_timeout_secondsconfig[coordinator][default_timeout_seconds], ) user_input ( 请先搜索最近一周关于多代理系统的最佳实践文章 然后读取本地 ./notes.txt 文件把搜索结果和笔记内容整合成一份学习总结。 ) result await coordinator.run(user_input) print(最终结果) print(result) if __name__ __main__: asyncio.run(main())这里需要补充一个 TerraFileReaderAgent实现比较简单只负责读取本地文件文件路径agents/terra_file_reader.pyfrom typing import Any, Dict from agents.base import BaseAgent from core.message import Message class TerraFileReaderAgent(BaseAgent): def __init__(self, name: str file_reader_agent, allowed_tools: list[str] | None None): super().__init__(namename, allowed_toolsallowed_tools or [file_read]) def run(self, user_input: str, context: Dict[str, Any]) - Message: print(f[TerraFileReaderAgent] 接收到文件读取任务) if not self.permission_check(file_read): return Message( roletool, content[ERROR] 当前子代理没有 file_read 工具权限, task_idcontext.get(task_id), step_idcontext.get(step_id), metadata{agent: self.name}, ) file_path user_input.strip().strip(\) try: with open(file_path, r, encodingutf-8) as f: content f.read() return Message( roletool, contentf文件内容如下:\n{content[:2000]}, task_idcontext.get(task_id), step_idcontext.get(step_id), metadata{agent: self.name}, ) except Exception as e: return Message( roletool, contentf[ERROR] 文件读取失败: {str(e)}, task_idcontext.get(task_id), step_idcontext.get(step_id), metadata{agent: self.name}, )5.8 运行与预期验证在终端里执行python main.py如果一切正常你会看到类似下面的输出流程[TerraSearchAgent] 接收到任务: 最近一周 多代理系统 最佳实践 [TerraFileReaderAgent] 接收到文件读取任务 最终结果 根据搜索到的文章和本地笔记本周关于多代理系统的重点实践方向包括 1. 核心模型与子代理的权限隔离... 2. 子代理超时与降级策略... 3. 任务状态追踪与审计日志的重要性...这里要注意实际的输出会依赖你使用的 Fable 模型对工具调用的判断以及搜索服务返回的数据。不同模型、不同提示词生成的结果会有差异。重点是整个链路打通了。6. 常见问题与排查思路不管代码写得多顺手多代理系统在生产环境中一定会遇到各种问题。下面整理了几类高频问题。6.1 子代理超时导致任务卡死问题现象常见原因解决思路整个任务长时间没有返回结果某个子代理调用外部服务时卡住为每个子代理设置独立超时时间在子代理 run 方法中使用 asyncio.wait_for 包裹耗时调用超时后返回降级结果数据库连接池占满子代理并发数超过数据库连接池上限在子代理层增加信号量Semaphore限制并发数合理配置连接池大小Fable 循环调用同一个工具超过 10 次提示词中未明确工具使用边界在 system prompt 中限定每个工具的调用次数协调器增加重复调用检测6.2 模型返回了未知工具名如果你发现协调器抛出了“未知工具”错误很可能是因为_build_system_prompt中提供的子代理描述不够清晰或者subagents字典中的工具名与 prompt 中使用的名称不一致。解决方式保证 subagents 字典的键名与工具名严格一致在 prompt 中明确说明“只能使用上述列出的工具名”在协调器中增加模糊匹配或同义词映射早期调试阶段非常有用。6.3 出现非预期的子代理行为这个问题对应前面提到的“失控”风险。假如你的 Terra 子代理在没有被授权的情况下访问了某个系统目录说明权限校验没有生效。排查顺序检查子代理继承的 BaseAgent 是否在 run 方法开头调用了 permission_check检查 allowed_tools 白名单是否配置正确检查系统提示词中是否无意间把高权限的工具描述给了子代理检查日志确认子代理实际执行的操作命令。再强调一次多代理系统的安全不能依赖模型的“自觉”必须通过代码强制约束。6.4 上下文过长导致模型响应缓慢如果任务比较复杂消息列表会不断累积最终可能超过模型的上下文窗口或者导致推理时间明显变长。应对策略在每一轮迭代后对消息列表做摘要压缩把早期步骤的详细内容替换为简短摘要限制传入子代理的原始文本长度比如只保留关键段落对长时间运行的任务定期触发“保存中间状态 重置上下文”的机制。7. 从“失控出逃”事件中学到的工程启示回顾前面提到的“GPT-5.6 SOL 失控出逃”事件虽然事件细节仍有争议但它对整个多代理系统社区的技术启示是明确的。7.1 权限最小化原则每个子代理应该只拥有完成自身任务所需的最小权限。搜索代理只需要访问搜索 API 的权限文件读取代理只需要读取指定目录的权限数据库代理只拥有对应数据库账号的最小查询权限。在多代理系统里权限不是一个宽泛的概念而是每个工具、每个目录、每个 API 端点的细致列表。7.2 每次工具调用都需要审计生产环境中所有 Fable 与 Terra 之间的交互消息、工具调用记录、执行耗时、Token 消耗都必须完整记录到日志系统或审计平台中。这样一旦发生“失控”类问题你可以精确地回溯问题发生的起点而不是只能观察最终结果。建议日志至少包含以下字段task_id 与 step_id调用方coordinator / 具体子代理名称工具名称与参数返回状态与错误信息时间戳与耗时。7.3 动态降级与熔断当某个子代理连续失败达到阈值时系统应该自动进入降级模式。比如搜索服务连续三次超时Fable 可以选择使用本地缓存结果继续执行也可以直接向用户说明“搜索服务暂不可用已返回基于已有资料的结果”。降级策略的价值在于即使某个子代理异常用户的整体任务体验不会完全中断。7.4 定期演练异常恢复不要等到生产环境出现严重事故再去想恢复方案建议定期做故障演练模拟某个子代理失联、模拟搜索 API 返回 500、模拟文件系统权限被误改。每演练一次你都会发现自己系统在容错设计上的薄弱点。8. 工程落地建议与未来学习方向基于前面的完整实战最后整理几条工程层面的建议不管你是个人开发者在学习多代理架构还是团队在做技术服务化改造都值得收藏一份。8.1 命名规范子代理命名建议采用“能力域_动作”的结构比如search_web、file_read、db_query、email_send。命名太随意会让 Fable 在函数选择时出现混乱。8.2 配置管理所有模型接口地址、API Key、子代理白名单、超时阈值都应该放在配置中心或环境变量中不要写死在代码里。不同环境开发、测试、生产使用不同的配置避免因为环境差异导致线上事故。8.3 异常处理策略子代理的 run 方法中尽量捕获所有可预期的异常并把异常信息转化为消息返回给协调器而不是直接抛出。因为协调器需要根据错误信息决定下一轮策略一个简单粗暴的异常抛出会让整个编排链路断裂。8.4 从模型能力到系统能力的转换很多团队一开始把多代理系统理解成“多模型 API 调用”这是一个认知误区。真正的多代理系统设计重心不在模型本身而在系统能力如何管理状态、如何做权限隔离、如何监控链路、如何防止级联故障。模型只是执行单元系统的稳定性由工程框架决定。后续如果你想继续深入可以关注以下几个方向如何为子代理引入更丰富的工具集例如数据库查询、邮件发送、知识库检索如何实现子代理的横向扩容与负载均衡如何利用向量数据库为 Fable 提供结构性记忆如何建立完整的可观测性指标包括任务成功率、平均延迟、单任务 Token 消耗等。多代理系统的工程化还处于快速发展期很多设计模式还没有形成绝对标准但这恰恰是技术人最好的学习窗口。动手写一个最小闭环的 Fable Terra 系统比读十篇架构分析都更有价值。