多代理架构在近两年的 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_url)
- Fable 对应的 API Key(api_key)
- Fable 对应的模型名称(model_name)
- Terra 对应的接口地址、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.py:Fable 核心编排器的实现。
- 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.yaml
fable: 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.py
from 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_factory=dict) 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.py
import 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(f"task {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.py
from 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.py
import 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__(name=name, allowed_tools=allowed_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( role="tool", content="[ERROR] 当前子代理没有 web_search 工具权限", task_id=context.get("task_id"), step_id=context.get("step_id"), metadata={"agent": self.name}, ) # 调用外部搜索服务 async with httpx.AsyncClient(timeout=10) 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( role="tool", content=f"搜索结果如下:\n{formatted}", task_id=context.get("task_id"), step_id=context.get("step_id"), metadata={"agent": self.name}, ) else: return Message( role="tool", content=f"[ERROR] 搜索服务返回状态码 {resp.status_code}", task_id=context.get("task_id"), step_id=context.get("step_id"), metadata={"agent": self.name}, )这个示例在搜索服务不可用时会返回错误信息,但不会把异常抛出,这是刻意为之——子代理调用外部服务失败时,应该把失败信息返回给核心模型,由核心模型决定是否更换策略。
5.6 实现 Fable 核心协调器
这是整个系统的核心文件。
文件路径:core/coordinator.py
from 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_url=coordination_model_config["base_url"], api_key=coordination_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(role="system", content=self._build_system_prompt(task_id)), Message(role="user", content=user_input, task_id=task_id), ] for iteration in range(self.max_iterations): # 调用 Fable 模型 response = await self.client.chat.completions.create( model=self.model_name, messages=[msg.to_dict() for msg in messages], temperature=0.2, ) assistant_content = response.choices[0].message.content or "" tool_calls = response.choices[0].message.tool_calls or [] messages.append(Message( role="assistant", content=assistant_content, task_id=task_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_input=arguments, context={ "task_id": task_id, "step_id": f"iter_{iteration}_tool_{tool_call.id}", }, ) messages.append(result_message) else: # 未知工具 messages.append(Message( role="tool", content=f"[ERROR] 未知工具: {func_name}", task_id=task_id, tool_call_id=tool_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.py
import 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( name="search_agent", allowed_tools=["web_search"], ) file_reader_agent = TerraFileReaderAgent( name="file_reader_agent", allowed_tools=["file_read"], ) subagents = { "search_agent": search_agent, "file_reader_agent": file_reader_agent, } coordinator = FableCoordinator( coordination_model_config=config["fable"], subagents=subagents, task_state_manager=task_state_manager, max_iterations=config["coordinator"]["max_iterations"], default_timeout_seconds=config["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.py
from 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__(name=name, allowed_tools=allowed_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( role="tool", content="[ERROR] 当前子代理没有 file_read 工具权限", task_id=context.get("task_id"), step_id=context.get("step_id"), metadata={"agent": self.name}, ) file_path = user_input.strip().strip("'\"") try: with open(file_path, "r", encoding="utf-8") as f: content = f.read() return Message( role="tool", content=f"文件内容如下:\n{content[:2000]}", task_id=context.get("task_id"), step_id=context.get("step_id"), metadata={"agent": self.name}, ) except Exception as e: return Message( role="tool", content=f"[ERROR] 文件读取失败: {str(e)}", task_id=context.get("task_id"), step_id=context.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 系统,比读十篇架构分析都更有价值。