实战指南：基于沙箱环境构建安全可控的多智能体系统

最近在尝试将大模型能力集成到实际业务中时，你是否也遇到过这样的困境：单个AI模型能力有限，处理复杂任务时逻辑混乱、成本高昂且难以控制？面对多步骤的客户咨询、自动化报告生成或代码审查等场景，我们需要的不是一个“万能”的模型，而是一个能协同工作、各司其职的智能体团队。这正是Harness工程与Multi-Agent（多智能体）系统要解决的核心问题。然而，网上资料要么过于学术化，要么就是零散的代码片段，缺乏一套从零搭建、安全可控且能直接复用于生产环境的完整指南。

本文将以实战为导向，为你系统拆解如何构建一个基于SandBox（沙箱）环境的多智能体系统。我们将从核心概念讲起，手把手带你完成环境搭建、智能体定义、任务编排与安全隔离，并最终实现一个能处理复杂任务的自动化流程。无论你是希望入门AI应用开发的工程师，还是寻求项目落地的技术负责人，这篇超过5000字的详尽教程都将提供清晰的路径和可运行的代码。

1. 背景与核心概念：为什么需要Harness与Multi-Agent？

在深入代码之前，我们必须厘清几个关键概念，理解它们为何成为当前AI工程化的热点。

1.1 AI大模型（Large Language Model, LLM）的局限与机遇当前的主流大模型（如GPT-4、Claude、DeepSeek等）在通用知识、逻辑推理和内容生成上表现出色。但它们存在固有瓶颈：单次交互的上下文长度有限、复杂任务容易产生“幻觉”、调用成本（按Token计价）高昂，且无法直接操作外部系统（如数据库、API）。这意味着，让一个大模型从头到尾独立完成一个包含数据查询、分析、撰写和发布的完整流程，既不可靠也不经济。

1.2 Agent（智能体）与Multi-Agent System（多智能体系统）一个Agent可以理解为一个具备特定技能、能感知环境、自主决策并执行动作的AI实体。它通常由三部分组成：

• 大脑（LLM）：负责理解指令、规划步骤、做出决策。
• 技能（Skill/Tool）：赋予Agent操作外部世界的能力，例如调用搜索API、运行Python代码、读写文件。
• 记忆（Memory）：保存对话历史、任务上下文，实现连贯交互。

而Multi-Agent System则由多个这样的智能体组成，它们通过协同、竞争或分层的方式共同完成一个复杂目标。例如，一个系统可能包含：

• Planner Agent（规划者）：拆解用户需求，生成任务执行流程图。
• Coder Agent（编码者）：专门编写和调试代码。
• Reviewer Agent（审查者）：检查代码质量和安全性。
• Executor Agent（执行者）：在安全环境中运行代码并返回结果。

这种分工协作的模式，显著提升了任务处理的可靠性、专业性和效率。

1.3 Harness（约束/驾驭）工程Harness在这里的含义是“约束”或“驾驭”。它的核心思想是：我们不能让AI智能体完全“自由发挥”，必须为其设计一套安全、可控的执行框架和环境。这包括：

• 资源隔离：防止智能体的代码操作影响宿主系统。
• 权限控制：限制智能体能访问的文件、网络和系统命令。
• 流程编排：定义智能体之间的通信协议和任务流转逻辑。
• 成本与性能监控：跟踪每个智能体的Token消耗和响应延迟。

Harness工程就是构建这套“缰绳”和“跑道”的实践，确保多智能体系统既能高效完成任务，又不会“脱缰”造成安全或稳定性问题。网络上热议的chimera项目探讨的正是异构LLM的低延迟与高性能服务，而actor-attention-critic则是多智能体强化学习中的一种协同策略。

1.4 SandBox（沙箱）—— 安全的执行环境SandBox是Harness工程中实现安全隔离的关键技术。它为每个智能体（尤其是需要执行代码的智能体）提供一个独立的、资源受限的运行环境。在这个“沙箱”里，智能体可以：

• 安全地执行Python、JavaScript等代码。
• 进行文件读写（限制在特定目录）。
• 访问网络（可配置白名单）。 即使代码存在恶意行为或错误，也不会危及主机系统。常见的沙箱技术包括Docker容器、gVisor、Firecracker以及一些语言级别的沙箱（如PyPy的沙盒模式）。

2. 环境准备与版本说明

我们将使用Python作为主要开发语言，因为它拥有最丰富的AI生态。本项目将模拟一个“数据分析与报告生成”的多智能体系统。

2.1 基础环境

• 操作系统：Ubuntu 20.04+/macOS Monterey+/Windows 10+ (WSL2推荐)。本文示例基于Ubuntu 22.04。
• Python版本：3.9 或 3.10。避免使用3.11+可能存在的某些库兼容性问题。
• 包管理工具：pip和venv(推荐) 或conda。

2.2 核心依赖库我们将创建一个requirements.txt文件来管理依赖。关键库及其作用如下：

# 核心AI与智能体框架 langchain==0.1.0 # 智能体编排框架，社区活跃 langchain-openai==0.0.5 # OpenAI模型集成 # 可选其他模型集成，如 langchain-anthropic, langchain-google-genai # 大模型API调用 (以OpenAI为例，可替换为其他) openai==1.12.0 # 代码执行与沙箱环境（核心！） docker==6.1.3 # 使用Docker作为沙箱后端 # 可选：piston-cli-client (纯Python沙箱)，但Docker更通用 # 工具与工具调用 langchain-experimental==0.0.49 # 包含一些实验性智能体，如代码执行智能体 requests==2.31.0 # 用于网络请求工具 # 实用工具 python-dotenv==1.0.0 # 管理环境变量（如API密钥） pydantic==2.5.0 # 数据验证

2.3 可选基础设施

• Docker & Docker Compose：用于构建和运行沙箱环境。这是实现安全代码执行的推荐方式。

  # Ubuntu 安装示例 sudo apt-get update sudo apt-get install docker.io docker-compose sudo usermod -aG docker $USER # 将当前用户加入docker组，避免sudo # 执行后需要**重新登录**生效

• 大模型API密钥：你需要准备一个或多个大模型的API密钥。本文示例使用OpenAI GPT-4，但你完全可以替换为DeepSeek、Claude或本地部署的Ollama模型。

2.4 项目结构预览在开始前，我们先规划好项目目录，这有助于理解后续代码的组织方式。

multi_agent_project/ ├── .env # 存储敏感信息，如API密钥 ├── requirements.txt # 项目依赖 ├── main.py # 主程序入口 ├── agents/ # 智能体模块目录 │ ├── __init__.py │ ├── base_agent.py # 智能体基类 │ ├── planner_agent.py # 规划智能体 │ ├── coder_agent.py # 编码智能体 │ └── critic_agent.py # 审查智能体 ├── tools/ # 工具（技能）目录 │ ├── __init__.py │ ├── code_executor.py # 代码执行工具（连接沙箱） │ └── web_search.py # 网络搜索工具（示例） ├── sandbox/ # 沙箱环境配置 │ ├── Dockerfile # 沙箱容器镜像定义 │ └── docker-compose.yml # 沙箱服务编排 ├── skills/ # 技能定义（LangChain Tool格式） │ └── __init__.py └── utils/ # 工具函数 └── __init__.py

3. 核心组件拆解：Agent, Tool, Sandbox

3.1 定义智能体（Agent）基类我们首先创建一个基础的智能体类，它封装了与大模型对话、调用工具的核心逻辑。使用LangChain可以大幅简化这一过程。

# agents/base_agent.py import os from typing import List, Optional, Any from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.tools import BaseTool from langchain.memory import ConversationBufferMemory from pydantic import BaseModel, Field class AgentConfig(BaseModel): """智能体配置模型""" name: str = Field(description="智能体名称") role: str = Field(description="智能体角色描述") llm_model: str = Field(default="gpt-4-turbo-preview", description="使用的LLM模型") temperature: float = Field(default=0.1, description="模型温度，越低越确定") tools: List[BaseTool] = Field(default_factory=list, description="可用的工具列表") system_prompt: str = Field(description="系统提示词，定义智能体行为") class BaseAgent: """智能体基类""" def __init__(self, config: AgentConfig): self.config = config self.llm = ChatOpenAI( model=config.llm_model, temperature=config.temperature, openai_api_key=os.getenv("OPENAI_API_KEY") # 从环境变量读取密钥 ) self.memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True) self.agent_executor = self._create_agent_executor() def _create_agent_executor(self) -> AgentExecutor: """创建LangChain智能体执行器""" prompt = ChatPromptTemplate.from_messages([ ("system", self.config.system_prompt), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), ]) agent = create_openai_tools_agent(self.llm, self.config.tools, prompt) return AgentExecutor(agent=agent, tools=self.config.tools, memory=self.memory, verbose=True) def run(self, task_input: str) -> str: """执行任务""" try: result = self.agent_executor.invoke({"input": task_input}) return result["output"] except Exception as e: return f"Agent {self.config.name} 执行出错: {str(e)}"

3.2 构建安全沙箱（Sandbox）安全是重中之重。我们将使用Docker创建一个轻量级、无网络（或受限网络）、只有必要Python包的沙箱环境。

# sandbox/Dockerfile FROM python:3.9-slim # 设置非root用户，增强安全性 RUN useradd -m -u 1000 sandboxuser && \ mkdir -p /app && chown -R sandboxuser:sandboxuser /app # 安装最小化依赖 COPY --chown=sandboxuser:sandboxuser requirements_sandbox.txt /app/requirements.txt RUN pip install --no-cache-dir -r /app/requirements.txt # 切换到非root用户 USER sandboxuser WORKDIR /app # 设置资源限制（在docker run时通过参数控制更灵活） # 这里主要进行环境隔离 # 启动命令：作为一个服务等待执行代码 CMD ["python", "-c", "import sys; print('Sandbox ready'); sys.stdout.flush(); import time; time.sleep(3600)"]

requirements_sandbox.txt只包含执行数据分析任务可能需要的包：

pandas==2.0.3 numpy==1.24.3 matplotlib==3.7.1 requests==2.31.0

然后，我们编写一个CodeExecutorTool，它负责将代码发送到Docker容器中执行。

# tools/code_executor.py import docker import os from typing import Optional from langchain.tools import BaseTool from pydantic import BaseModel, Field class CodeExecutionInput(BaseModel): code: str = Field(description="需要执行的Python代码字符串") timeout: int = Field(default=30, description="执行超时时间（秒）") class CodeExecutorTool(BaseTool): name = "code_executor" description = "在安全的Docker沙箱中执行Python代码并返回结果。适用于数据分析、计算和文件操作。" args_schema = type("CodeExecutionInput", (BaseModel,), { "code": Field(..., description="需要执行的Python代码字符串"), "timeout": Field(default=30, description="执行超时时间（秒）") }) def __init__(self): super().__init__() # 初始化Docker客户端 self.client = docker.from_env() self.container_image = "my_sandbox:latest" # 需提前构建的镜像 self.container = None self._start_container() def _start_container(self): """启动或复用沙箱容器""" try: # 查找正在运行的沙箱容器 containers = self.client.containers.list(filters={"ancestor": self.container_image}) if containers: self.container = containers[0] print(f"复用现有容器: {self.container.id[:12]}") else: # 启动新容器，限制资源并禁用网络 self.container = self.client.containers.run( self.container_image, detach=True, network_mode='none', # 禁用网络访问 mem_limit='512m', # 内存限制 pids_limit=50, # 进程数限制 cpu_period=100000, cpu_quota=50000, # 限制CPU使用率 volumes={'/tmp/sandbox': {'bind': '/app/data', 'mode': 'rw'}} # 仅挂载数据卷 ) print(f"启动新容器: {self.container.id[:12]}") except docker.errors.ImageNotFound: raise RuntimeError(f"镜像 {self.container_image} 未找到，请先构建。") def _run(self, code: str, timeout: int = 30) -> str: """核心执行方法""" if not self.container: return "错误：沙箱容器未就绪。" exec_command = f"python -c {repr(code)}" try: exec_result = self.container.exec_run( cmd=["python", "-c", code], workdir="/app/data", timeout=timeout ) exit_code, output = exec_result.exit_code, exec_result.output.decode('utf-8') if exit_code == 0: return output.strip() if output else "代码执行成功，无输出。" else: return f"代码执行错误 (Exit Code: {exit_code}):\n{output}" except docker.errors.APIError as e: return f"Docker API错误: {str(e)}" except Exception as e: return f"执行异常: {str(e)}" def _arun(self, code: str, timeout: int = 30): raise NotImplementedError("此工具不支持异步执行。")

3.3 设计技能（Skill/Tool）技能是智能体的“手”和“脚”。除了代码执行，我们还可以定义其他技能，如网络搜索。

# tools/web_search.py import requests from langchain.tools import BaseTool from pydantic import BaseModel, Field import json class WebSearchInput(BaseModel): query: str = Field(description="搜索查询词") class WebSearchTool(BaseTool): name = "web_search" description = "使用搜索引擎API获取最新信息。输入是一个搜索查询字符串。" args_schema = type("WebSearchInput", (BaseModel,), { "query": Field(..., description="搜索查询词") }) def __init__(self, api_key: str): super().__init__() # 示例：使用Serper Dev API (免费额度) self.api_key = api_key self.url = "https://google.serper.dev/search" def _run(self, query: str) -> str: headers = { 'X-API-KEY': self.api_key, 'Content-Type': 'application/json' } payload = json.dumps({"q": query}) try: response = requests.post(self.url, headers=headers, data=payload, timeout=10) response.raise_for_status() results = response.json() # 简化返回前3个结果 simplified = [] for item in results.get('organic', [])[:3]: simplified.append({ "title": item.get('title'), "link": item.get('link'), "snippet": item.get('snippet') }) return json.dumps(simplified, ensure_ascii=False, indent=2) except requests.exceptions.RequestException as e: return f"网络搜索失败: {str(e)}" def _arun(self, query: str): raise NotImplementedError("此工具不支持异步执行。")

4. 完整实战案例：构建数据分析报告生成系统

现在，我们将上述组件组装起来，实现一个包含三个智能体的系统：规划者(Planner)、编码者(Coder)、审查者(Critic)。它们将协作完成“分析某公司销售数据并生成报告”的任务。

4.1 定义具体的智能体

# agents/planner_agent.py from .base_agent import BaseAgent, AgentConfig from langchain.tools import Tool class PlannerAgent(BaseAgent): """规划智能体：负责拆解任务，生成执行步骤""" def __init__(self): config = AgentConfig( name="Planner", role="你是一个资深项目分析师，擅长将模糊的需求拆解为具体、可执行的数据分析步骤。", llm_model="gpt-4-turbo-preview", temperature=0.1, tools=[], # 规划者通常不需要外部工具 system_prompt="""你负责规划数据分析任务。用户会提出一个需求，你需要将其拆解成一个清晰的、线性的步骤列表。 每个步骤应该明确指定由哪个智能体执行（Coder或Critic），并描述该步骤的具体输入和预期输出。 输出格式必须是严格的JSON列表，例如： [ {"step": 1, "agent": "Coder", "action": "生成模拟销售数据", "input": "创建包含日期、产品、销售额、数量的DataFrame"}, {"step": 2, "agent": "Coder", "action": "计算月度销售额趋势", "input": "对数据按月份聚合，计算总销售额"}, {"step": 3, "agent": "Critic", "action": "检查数据合理性", "input": "审查上一步生成的趋势图和数据，确保没有异常值"} ] 只输出JSON，不要有其他解释。""" ) super().__init__(config) # agents/coder_agent.py from .base_agent import BaseAgent, AgentConfig from tools.code_executor import CodeExecutorTool from tools.web_search import WebSearchTool import os class CoderAgent(BaseAgent): """编码智能体：负责编写和执行Python代码来完成数据任务""" def __init__(self): # 初始化工具 code_tool = CodeExecutorTool() search_tool = WebSearchTool(api_key=os.getenv("SERPER_API_KEY", "")) config = AgentConfig( name="Coder", role="你是一个专业的Python数据分析师，精通pandas, numpy, matplotlib。根据指令编写安全、高效的代码并在沙箱中执行。", llm_model="gpt-4-turbo-preview", temperature=0.2, tools=[code_tool, search_tool], system_prompt="""你是一个Python代码专家。你的任务是接收具体的操作描述（如‘生成模拟销售数据’），然后编写出能完成该任务的Python代码，并使用`code_executor`工具执行它。 编写代码时请遵循以下规则： 1. 代码必须完整、可独立运行。 2. 优先使用pandas进行数据处理。 3. 如果需要可视化，使用matplotlib生成图表并保存为PNG文件到`/app/data/`目录。 4. 如果任务需要最新信息（如‘查找2023年智能手机市场趋势’），可以使用`web_search`工具。 5. 最终输出应包括：代码简要说明、执行结果（如数据预览、图表保存路径）或关键发现。 如果执行出错，分析错误并尝试修复。""" ) super().__init__(config) # agents/critic_agent.py from .base_agent import BaseAgent, AgentConfig from tools.code_executor import CodeExecutorTool class CriticAgent(BaseAgent): """审查智能体：负责检查代码质量、结果合理性和安全性""" def __init__(self): code_tool = CodeExecutorTool() # 用于重新运行代码验证 config = AgentConfig( name="Critic", role="你是一个严谨的代码审查员和数据质量专家。你的任务是发现代码缺陷、逻辑错误和不合理的结果。", llm_model="gpt-4-turbo-preview", temperature=0.1, tools=[code_tool], system_prompt="""你负责审查Coder智能体生成的代码和结果。你的输入将包含：1) 任务描述，2) 已编写的代码，3) 代码执行结果。 你需要仔细检查： 1. **安全性**：代码是否尝试执行危险操作（如删除系统文件、无限循环）？ 2. **正确性**：代码逻辑是否符合任务要求？是否存在语法或运行时错误？ 3. **数据合理性**：结果数据是否有异常值（如负的销售额）？趋势是否符合常识？ 4. **代码质量**：是否有冗余计算？变量命名是否清晰？ 输出你的审查意见，明确指出任何问题，并给出修改建议。如果一切正常，输出‘审查通过：代码与结果均符合预期’。""" ) super().__init__(config)

4.2 主控流程与任务编排这是整个系统的大脑，负责协调各个智能体按顺序工作。

# main.py import os import json from agents.planner_agent import PlannerAgent from agents.coder_agent import CoderAgent from agents.critic_agent import CriticAgent from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 class MultiAgentOrchestrator: """多智能体编排器""" def __init__(self): self.planner = PlannerAgent() self.coder = CoderAgent() self.critic = CriticAgent() self.execution_log = [] def execute_plan(self, plan: list): """按计划执行每一步""" results = {} for step in plan: step_num = step['step'] agent_name = step['agent'] action = step['action'] task_input = step.get('input', '') print(f"\n{'='*50}") print(f"执行步骤 {step_num}: [{agent_name}] - {action}") print(f"输入: {task_input}") agent = getattr(self, agent_name.lower(), None) if not agent: result = f"错误：未知的智能体 {agent_name}" print(result) results[step_num] = result continue # 执行任务 try: # 对于Coder，输入是动作描述；对于Critic，输入需要包含上下文（上一步的结果） if agent_name == "Critic" and step_num > 1: # 将上一步的结果作为审查的上下文 prev_result = results.get(step_num-1, "") full_input = f"任务：{action}\n待审查的代码及结果：\n{prev_result}" else: full_input = task_input output = agent.run(full_input) print(f"输出:\n{output}") results[step_num] = output self.execution_log.append({ "step": step_num, "agent": agent_name, "input": task_input, "output": output[:500] # 截断长输出 }) except Exception as e: error_msg = f"步骤 {step_num} 执行异常: {str(e)}" print(error_msg) results[step_num] = error_msg return results def run(self, user_request: str): """运行完整流程：规划 -> 执行 -> 总结""" print(f"用户请求: {user_request}") print("\n[阶段1] 规划任务...") plan_json_str = self.planner.run(user_request) print(f"生成的计划:\n{plan_json_str}") try: # 解析规划结果 plan = json.loads(plan_json_str.strip()) if not isinstance(plan, list): raise ValueError("规划结果不是有效的JSON列表") except json.JSONDecodeError as e: print(f"解析计划失败: {e}") plan = [{"step": 1, "agent": "Coder", "action": "直接处理请求", "input": user_request}] print("\n[阶段2] 执行计划...") results = self.execute_plan(plan) print("\n[阶段3] 生成最终报告...") # 可以在这里添加一个ReportAgent来汇总结果，此处简化为打印 final_report = f""" ========== 任务执行完成 ========== 原始请求: {user_request} 执行步骤数: {len(plan)} 最终结果摘要: {json.dumps(results, indent=2, ensure_ascii=False)} """ print(final_report) return final_report if __name__ == "__main__": # 检查必要环境变量 if not os.getenv("OPENAI_API_KEY"): print("错误：请在 .env 文件中设置 OPENAI_API_KEY") exit(1) orchestrator = MultiAgentOrchestrator() # 示例任务 user_request = "分析一下我们公司过去一年的销售数据，找出最畅销的产品类别和销售额的季节性趋势，并生成一份简要报告。" # 注意：实际任务中，Coder需要真实数据。这里规划器会指示Coder先‘生成模拟数据’。 orchestrator.run(user_request)

4.3 运行与验证

1. 构建沙箱镜像：在项目根目录下，执行docker build -t my_sandbox:latest -f sandbox/Dockerfile .
2. 设置环境变量：创建.env文件，填入你的API密钥。

   OPENAI_API_KEY=sk-your-openai-key-here SERPER_API_KEY=your-serper-key-here # 可选，用于搜索

3. 安装依赖：pip install -r requirements.txt
4. 运行主程序：python main.py

4.4 预期输出与结果说明程序运行后，你将在控制台看到类似以下的输出，清晰地展示了多智能体协作的整个过程：

用户请求: 分析一下我们公司过去一年的销售数据... [阶段1] 规划任务... 生成的计划: [ {"step": 1, "agent": "Coder", "action": "生成模拟的过去一年销售数据", "input": "创建包含日期、产品类别、销售额、数量的DataFrame，时间范围从2023-01-01至今"}, {"step": 2, "agent": "Coder", "action": "计算各产品类别的总销售额，找出最畅销类别", "input": "对模拟数据按产品类别分组，汇总销售额"}, ... ] [阶段2] 执行计划... ================================================== 执行步骤 1: [Coder] - 生成模拟的过去一年销售数据 输入: 创建包含日期、产品类别、销售额、数量的DataFrame... （Coder开始思考并调用code_executor工具） > 进入链... > 调用工具: code_executor... 工具输出: 代码执行成功。已生成包含1000条记录的DataFrame。前5行预览：... > 链结束。 输出: 已成功生成模拟销售数据... ================================================== 执行步骤 2: [Coder] - 计算各产品类别的总销售额... ... [阶段3] 生成最终报告... ========== 任务执行完成 ========== ...

最终，在沙箱的挂载目录/tmp/sandbox（根据docker-compose.yml配置）中，你可能会找到由Coder智能体生成的图表文件（如monthly_trend.png）。

5. 常见问题与排查思路

在搭建和运行多智能体系统时，你可能会遇到以下典型问题：

6. 最佳实践与工程建议

将多智能体系统投入生产环境或复杂项目时，以下经验能帮助你走得更稳。

6.1 智能体设计原则

• 单一职责：每个智能体应专注于一个明确的领域（如规划、编码、审查、搜索）。这能提升可靠性和可调试性。
• 清晰的边界与协议：明确定义智能体之间的通信数据格式（如使用JSON Schema）。规划器的输出必须是结构化的，便于主控制器解析。
• 人机协同与审批点：在关键步骤（如删除数据、发布结果）设置人工审批环节，避免AI完全自主做出不可逆的操作。

6.2 沙箱安全强化

• 最小权限原则：沙箱容器应以非root用户运行，并移除所有不必要的系统权限（--cap-drop=ALL）。
• 资源硬限制：通过Docker的--memory,--cpus,--pids-limit等参数严格限制CPU、内存和进程数，防止资源耗尽攻击。
• 文件系统隔离：仅挂载必要的只读或特定读写卷。避免挂载主机敏感目录。
• 网络隔离：默认禁用网络（--network=none）。对于需要网络访问的工具（如搜索），应创建独立的、仅有出站权限的网络沙箱，或使用代理。

6.3 性能与成本优化

• 智能体路由：并非所有任务都需要最强的GPT-4。可以根据任务复杂度路由到不同模型（如简单分类用GPT-3.5-Turbo，复杂推理用GPT-4），以平衡成本与效果。
• 缓存与记忆：对频繁出现的相似查询结果进行缓存。利用LangChain的ConversationSummaryMemory或VectorStoreRetrieverMemory来管理长对话上下文，而非无限制地增长Token消耗。
• 异步执行：如果智能体之间的任务没有强依赖，可以使用异步框架（如asyncio）并发执行，减少总体延迟。

6.4 可观测性与监控

• 全面日志记录：记录每个智能体的输入、输出、调用的工具、消耗的Token数以及执行时间。self.execution_log只是一个开始，应集成到ELK或Prometheus+Grafana中。
• Token成本跟踪：在调用LLM API时，捕获返回中的usage字段，实时统计和预警成本。
• 链路追踪：为每个用户会话或任务生成唯一ID，贯穿所有智能体和工具调用，便于问题排查和流程分析。

6.5 配置与版本管理

• 配置外部化：将模型类型、API端点、温度参数、沙箱配置等全部移至配置文件（如config.yaml）或环境变量，便于不同环境（开发、测试、生产）切换。
• 依赖锁定：使用pip-tools或poetry精确锁定所有依赖库的版本，确保环境一致性。
• 容器镜像版本化：沙箱的Docker镜像应打上版本标签，并与主应用代码版本关联，确保每次部署的环境完全相同。

通过本教程，你不仅学会了如何搭建一个多智能体系统的原型，更重要的是掌握了其核心设计思想：通过Harness工程（沙箱、编排、约束）将强大的但不可控的AI能力，转化为安全、可靠、可协作的生产力组件。下一步，你可以尝试集成更多的智能体（如专精SQL查询的、专精邮件撰写的），或者将其与你的实际业务系统（如CRM、数据库）对接，解决真实的自动化需求。记住，从一个小而具体的场景开始迭代，远比一开始就设计一个庞大复杂的系统要高效得多。