在实际 AI 项目开发中，我们常常面临一个困境：一个强大的基础模型（如 Gemini）虽然能力出众，但直接用于解决特定业务问题时，往往需要大量的提示工程、上下文管理和任务编排。这就像拥有一台顶级发动机，却需要手动组装变速箱、传动轴和方向盘才能造出一辆能上路的车。近年来，智能体（Agent）架构为解决这一问题提供了新思路，它让模型能够自主规划、使用工具并执行复杂任务链。而“智能体协作”则将这一思路推向更高维度，通过多个智能体的分工与配合，处理更庞大、更动态的问题。

本文将以“训练一个机器人模型”作为目标场景，探讨如何利用 Gemini 3.7 Flash 这类高效模型作为核心“大脑”，构建一个能够自主协作的智能体系统来完成此任务。我们将从零开始，设计一个包含规划、执行、评估等多个角色的智能体协作框架，并实现一个可以自动执行数据收集、模型训练、评估反馈等步骤的自主训练流程。无论你是希望深入理解智能体架构的开发者，还是正在寻找自动化 AI 工作流解决方案的工程师，这篇文章都将提供一个从概念到代码实现的完整路径。

1. 理解智能体协作与自主训练的核心概念

在开始构建之前，我们需要厘清几个关键概念，这决定了我们后续架构设计的方向。

1.1 什么是智能体（Agent）？

在 AI 语境下，智能体远不止是一个调用 API 的脚本。它是一个具备一定自主性的系统，通常包含几个核心组件：

• 规划（Planning）：将大目标分解为可执行的子任务序列。
• 工具使用（Tool Use）：能够调用外部函数、API 或执行代码来获取信息或改变环境状态。
• 记忆（Memory）：保存对话历史、任务上下文和执行结果，用于后续决策。
• 执行（Execution）：根据规划调用模型或工具，并处理返回结果。

一个简单的聊天机器人不是智能体，但一个能根据你“分析上周销售数据并生成报告”的指令，自动登录数据库、查询、分析并调用图表生成工具的 AI 程序，就是一个初级智能体。

1.2 为什么需要智能体“协作”？

单个智能体的能力受限于其提示词、工具集和上下文长度。对于“训练一个机器人模型”这样的复杂任务，涉及步骤繁多（环境搭建、数据标注、训练脚本调试、超参数调整、评估部署），且可能需要在不同专业领域（如 Python 编程、Shell 命令、结果评估）间切换。让一个智能体完成所有工作，极易导致任务规划混乱、上下文过载和错误累积。

智能体协作通过角色分工来解决这个问题：

• 规划者（Planner）：负责顶层任务分解和流程控制。它理解最终目标，并制定分阶段计划。
• 执行者（Executor）：负责具体操作。它接收规划者的子任务指令，调用相应的工具（如运行 Python 脚本、执行 Shell 命令）来完成任务。
• 评估者（Evaluator）：负责质量检查。它分析执行结果（如训练日志、评估指标），判断任务是否成功，并向规划者提供反馈。

多个智能体通过共享的工作区或消息总线进行通信，形成一个自主的协作系统。

1.3 自主训练机器人模型意味着什么？

“自主训练”在此处指代一个高度自动化的流程，系统在给定高层目标（如“训练一个能在模拟环境中移动的机器人模型”）和必要资源（如模拟器、计算环境）后，能够自行完成以下循环：

1. 分析需求：理解任务目标和技术约束（如使用 PyTorch 还是 TensorFlow，是否需要特定传感器数据）。
2. 生成计划：创建数据准备、模型选择、训练、验证的步骤。
3. 执行代码：编写或调用现有的数据预处理、模型定义和训练脚本。
4. 监控与调整：运行训练，监控损失函数和评估指标，根据情况调整超参数或重新规划。
5. 输出结果：保存训练好的模型，并生成训练报告。

这个过程的核心挑战在于如何将非结构化的高级目标，转化为一系列可被可靠执行的结构化操作（代码、命令），并处理执行过程中出现的各种异常（如库缺失、脚本错误、训练发散）。

2. 环境准备与核心工具选型

要实现上述构想，我们需要搭建一个开发环境，并选择一系列能够支撑智能体协作的工具。Gemini 3.7 Flash 模型将作为我们所有智能体的“思考核心”。

2.1 基础开发环境配置

首先，确保你有一个可用的 Python 环境（推荐 3.9 以上版本）和基本的开发工具。

# 1. 创建并激活一个独立的 Python 虚拟环境（强烈推荐） python -m venv agent_workspace source agent_workspace/bin/activate # Linux/macOS # 或 agent_workspace\Scripts\activate # Windows # 2. 升级包管理工具 pip install --upgrade pip # 3. 安装 Jupyter Lab 或 Notebook（用于交互式开发和调试） pip install jupyterlab

2.2 核心依赖安装：智能体框架与 Gemini SDK

我们将使用LangChain作为智能体框架的基础，因为它提供了成熟的智能体、工具和记忆体抽象。同时，安装 Google 的google-generativeai库来调用 Gemini 模型。

# 安装 LangChain 及其社区工具包 pip install langchain langchain-community langchain-google-genai # 安装 Gemini SDK pip install google-generativeai # 安装一些可能用到的工具依赖 pip install requests python-dotenv # 用于网络请求和环境变量管理

2.3 获取并配置 Gemini API 密钥

1. 访问 Google AI Studio (https://aistudio.google.com/)。
2. 登录你的 Google 账号。
3. 在界面中，你应该可以找到创建 API 密钥的选项。生成一个密钥并复制。
4. 在项目根目录创建一个名为.env的文件，将密钥存入：

# .env 文件内容 GOOGLE_API_KEY=你的_实际_API_密钥_在这里

重要安全提示：务必在.gitignore文件中加入.env，切勿将 API 密钥提交到版本控制系统。

2.4 辅助工具准备：模拟训练环境

为了演示“训练机器人模型”，我们需要一个简单的训练任务和环境。这里我们选择gym（OpenAI Gym）和stable-baselines3这个强化学习库，它们能让我们快速搭建一个机器人（或智能体）在模拟环境中学习的案例。这纯粹为了提供具体的、可执行的训练目标。

# 安装强化学习模拟环境和库 pip install gym==0.26.2 # 注意版本兼容性 pip install stable-baselines3[extra] pip install pyglet<=1.5.27 # 解决某些渲染兼容性问题

如果只是关注智能体协作流程，不运行具体训练，这部分依赖可选。

3. 构建基于 Gemini 的智能体协作系统

现在，我们开始构建系统的核心。我们将创建三个智能体：规划者、执行者、评估者，并设计它们之间的协作机制。

3.1 项目结构设计

一个清晰的项目结构有助于管理复杂度。

gemini_agent_trainer/ ├── .env # 环境变量（API密钥） ├── requirements.txt # 项目依赖 ├── main.py # 主程序入口 ├── agents/ # 智能体模块目录 │ ├── __init__.py │ ├── planner_agent.py # 规划者智能体 │ ├── executor_agent.py # 执行者智能体 │ └── evaluator_agent.py # 评估者智能体 ├── tools/ # 工具定义目录 │ ├── __init__.py │ ├── code_tools.py # 代码读写执行工具 │ ├── shell_tools.py # Shell命令工具 │ └── training_tools.py # 训练相关工具 ├── workspace/ # 共享工作区 │ ├── plans/ # 存放生成的计划 │ ├── scripts/ # 存放生成的代码脚本 │ └── logs/ # 存放执行日志和评估结果 └── utils/ # 工具函数 ├── __init__.py └── message_bus.py # 简单的消息传递模拟

3.2 实现核心工具：智能体的“手和脚”

智能体通过工具与世界交互。我们先实现几个关键工具。

文件读写与代码执行工具 (tools/code_tools.py):

import subprocess import sys from typing import Type 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 PythonCodeTool(BaseTool): name = "execute_python_code" description = "执行一段Python代码并返回结果或错误。用于数据计算、测试脚本等。" args_schema: Type[BaseModel] = CodeExecutionInput def _run(self, code: str, timeout: int = 30) -> str: try: # 使用 subprocess 在独立进程中运行代码，更安全 result = subprocess.run( [sys.executable, "-c", code], capture_output=True, text=True, timeout=timeout ) if result.returncode == 0: return f"执行成功。输出：\n{result.stdout}" else: return f"执行失败。错误：\n{result.stderr}" except subprocess.TimeoutExpired: return f"错误：代码执行超时（>{timeout}秒）。" except Exception as e: return f"工具内部错误：{str(e)}" async def _arun(self, code: str, timeout: int = 30): # 异步支持，此处简化 raise NotImplementedError("此工具暂不支持异步执行") # 类似地，可以创建 FileReadTool, FileWriteTool 等

Shell 命令执行工具 (tools/shell_tools.py):

import subprocess import os from typing import Type from langchain.tools import BaseTool from pydantic import BaseModel, Field class ShellCommandInput(BaseModel): """Shell命令工具的输入模型。""" command: str = Field(description="要执行的Shell命令") work_dir: str = Field(default=".", description="执行命令的工作目录") class ShellCommandTool(BaseTool): name = "execute_shell_command" description = "在指定工作目录下执行一条Shell命令（如 ls, pip install, python script.py）。请谨慎使用危险命令。" args_schema: Type[BaseModel] = ShellCommandInput def _run(self, command: str, work_dir: str = ".") -> str: # 简单安全检查（可根据需要扩展） dangerous_keywords = ["rm -rf /", "format", ":(){:|:&};:"] for kw in dangerous_keywords: if kw in command: return f"拒绝执行可能危险的命令：{command}" try: original_dir = os.getcwd() os.chdir(work_dir) result = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=60 ) os.chdir(original_dir) output = f"命令执行完毕。返回码：{result.returncode}\n" if result.stdout: output += f"标准输出：\n{result.stdout}\n" if result.stderr: output += f"标准错误：\n{result.stderr}" return output except subprocess.TimeoutExpired: return f"错误：命令执行超时。" except Exception as e: return f"工具内部错误：{str(e)}"

3.3 构建智能体：规划者、执行者、评估者

规划者智能体 (agents/planner_agent.py):规划者的核心是理解终极目标，并产出结构化的计划。我们使用 LangChain 的create_react_agent模式，并赋予它“思考”的能力。

import os from langchain.agents import create_react_agent, AgentExecutor from langchain_google_genai import ChatGoogleGenerativeAI from langchain.prompts import PromptTemplate from langchain.tools import Tool from tools.code_tools import PythonCodeTool from tools.shell_tools import ShellCommandTool # 初始化 Gemini 模型 llm = ChatGoogleGenerativeAI( model="gemini-1.5-flash", # 或 gemini-1.5-pro，根据可用性调整 temperature=0.1, # 低温度保证计划更确定、可重复 google_api_key=os.getenv("GOOGLE_API_KEY") ) # 规划者可以使用的工具（主要是信息获取和简单验证） planning_tools = [ Tool( name="Inspect Directory", func=lambda dir: str(os.listdir(dir)), description="列出指定目录下的文件和文件夹。用于了解工作区状态。" ), # 可以添加更多信息查询工具 ] # 规划者专用提示词 PLANNER_PROMPT = PromptTemplate.from_template(""" 你是一个高级任务规划AI。你的目标是将用户的高层目标分解成一个详细、可执行、线性的步骤计划。 当前工作区状态： {workspace_status} 用户目标：{input} 请生成一个JSON格式的计划，包含以下字段： - `overall_goal`: 重申总体目标。 - `steps`: 一个步骤对象数组，每个对象包含： - `id`: 步骤序号。 - `description`: 该步骤的详细描述。 - `agent`: 负责执行此步骤的智能体类型 (`executor` 或 `evaluator`)。 - `deliverable`: 该步骤应产生的交付物（如文件、日志、指标）。 - `dependencies`: 此步骤依赖的前置步骤ID列表。 请确保计划逻辑清晰，步骤粒度适中，一个步骤通常对应执行者的一次主要工具调用或评估者的一次分析。 只输出JSON，不要有其他解释。 """) def create_planner_agent(): # 使用 ReAct 范式创建规划者 planner = create_react_agent( llm=llm, tools=planning_tools, prompt=PLANNER_PROMPT ) planner_executor = AgentExecutor( agent=planner, tools=planning_tools, handle_parsing_errors=True, verbose=True # 调试时打开，生产环境关闭 ) return planner_executor if __name__ == "__main__": # 简单测试 import dotenv dotenv.load_dotenv() planner = create_planner_agent() result = planner.invoke({ "input": "训练一个能在CartPole-v1环境中保持平衡的强化学习模型。", "workspace_status": "工作区为空。" }) print(result["output"])

执行者与评估者智能体：执行者和评估者结构类似，但提示词和工具集不同。执行者拥有强大的操作工具（代码执行、Shell命令），而评估者拥有分析工具（读取日志、计算指标、对比结果）。

由于篇幅限制，这里给出执行者智能体的核心提示词思路：

EXECUTOR_PROMPT_TEMPLATE = """ 你是一个AI执行者，擅长使用工具精确完成具体任务。 当前任务：{current_step} 你有以下工具：{tools} 任务历史上下文：{agent_scratchpad} 请严格按照任务要求，选择合适的工具并执行。如果任务需要多步操作，请逐步进行。 如果遇到错误，分析错误信息并尝试修复，如果无法修复，请明确报告失败原因。 你的输出应该是工具执行的结果，或者是最终的任务完成状态报告。 """

评估者智能体的提示词则侧重于分析：

EVALUATOR_PROMPT_TEMPLATE = """ 你是一个AI评估者，负责分析任务执行结果的质量。 需要评估的交付物：{deliverable} 评估标准：{criteria} （例如：训练损失下降、验证准确率>90%、脚本无语法错误） 执行日志或结果：{execution_result} 请分析结果是否满足标准。如果满足，给出“SUCCESS”结论和简要总结。 如果不满足，给出“FAILURE”结论，详细说明哪里不符合标准，并可能的原因分析。 你的输出将用于决定任务流程是继续、重试还是终止。 """

3.4 实现协作引擎：主控循环

这是整个系统的大脑，负责初始化智能体、调用规划者生成计划、按顺序调度执行者和评估者，并处理步骤间的依赖关系。

# main.py 核心部分 import json import os from agents.planner_agent import create_planner_agent # 假设我们已经实现了 create_executor_agent 和 create_evaluator_agent from agents.executor_agent import create_executor_agent from agents.evaluator_agent import create_evaluator_agent class CollaborativeTrainingOrchestrator: def __init__(self): self.planner = create_planner_agent() self.executor = create_executor_agent() self.evaluator = create_evaluator_agent() self.workspace_status = "初始状态" self.plan = None def generate_plan(self, user_goal: str): """调用规划者生成计划""" print(f"[Orchestrator] 正在为目标生成计划：{user_goal}") result = self.planner.invoke({ "input": user_goal, "workspace_status": self.workspace_status }) try: # 解析规划者输出的JSON plan_json = json.loads(result["output"]) self.plan = plan_json print(f"[Orchestrator] 计划生成成功，共 {len(plan_json['steps'])} 个步骤。") # 将计划保存到工作区 plan_file = f"workspace/plans/plan_{hash(user_goal)}.json" os.makedirs(os.path.dirname(plan_file), exist_ok=True) with open(plan_file, 'w') as f: json.dump(plan_json, f, indent=2) return plan_json except json.JSONDecodeError as e: print(f"[Orchestrator] 错误：无法解析规划者输出。原始输出：{result['output']}") return None def execute_step(self, step): """执行单个步骤，根据步骤类型分发给执行者或评估者""" step_id = step["id"] description = step["description"] agent_type = step.get("agent", "executor") print(f"\n=== 执行步骤 {step_id} ===") print(f"描述：{description}") print(f"执行者：{agent_type}") if agent_type == "executor": result = self.executor.invoke({"current_step": description}) elif agent_type == "evaluator": # 评估者需要更多上下文，如交付物和标准 deliverable = step.get("deliverable", "") criteria = step.get("criteria", "任务完成且无报错") # 这里需要从工作区获取实际的执行结果，此处简化 execution_result = f"结果待填充，来自步骤{step_id}" result = self.evaluator.invoke({ "deliverable": deliverable, "criteria": criteria, "execution_result": execution_result }) else: result = {"output": f"错误：未知的智能体类型 '{agent_type}'"} print(f"步骤 {step_id} 结果：{result['output'][:200]}...") # 截断输出 # 更新工作区状态（例如，记录步骤完成情况、保存产出文件） self.workspace_status += f"\n步骤{step_id}({agent_type})完成：{description}" return result def run(self, user_goal: str): """主运行循环""" plan = self.generate_plan(user_goal) if not plan: print("计划生成失败，流程终止。") return completed_steps = set() steps = plan["steps"] while len(completed_steps) < len(steps): progress_made = False for step in steps: step_id = step["id"] if step_id in completed_steps: continue # 检查依赖是否全部满足 dependencies = step.get("dependencies", []) if all(dep in completed_steps for dep in dependencies): result = self.execute_step(step) # 简单判断成功（实际应根据评估者结论） if "SUCCESS" in result["output"] or "失败" not in result["output"]: completed_steps.add(step_id) progress_made = True else: print(f"步骤 {step_id} 执行失败，流程暂停。") # 这里可以加入重试或人工干预逻辑 return else: # 依赖未满足，跳过 continue if not progress_made: print("检测到死锁或循环依赖，流程终止。") break print(f"\n[Orchestrator] 所有步骤完成！目标 '{user_goal}' 处理完毕。") print(f"最终工作区状态：\n{self.workspace_status}") if __name__ == "__main__": import dotenv dotenv.load_dotenv() orchestrator = CollaborativeTrainingOrchestrator() # 启动协作训练流程 orchestrator.run("训练一个能在CartPole-v1环境中保持平衡的强化学习模型，使用PPO算法，训练10000步，并保存模型。")

4. 运行验证与结果分析

4.1 启动协作训练流程

确保所有文件就位，并在项目根目录下运行主程序：

cd /path/to/gemini_agent_trainer python main.py

如果一切配置正确，你将看到类似以下的输出（具体内容因 Gemini 生成结果而异）：

[Orchestrator] 正在为目标生成计划：训练一个能在CartPole-v1环境中保持平衡的强化学习模型... [Orchestrator] 计划生成成功，共 6 个步骤。 === 执行步骤 1 === 描述：检查当前Python环境并安装必要的依赖（gym, stable-baselines3）。 执行者：executor > 进入新的AgentExecutor链... 思考：我需要使用Shell命令工具来安装Python包。 行动：execute_shell_command 行动输入：{"command": "pip install gym stable-baselines3", "work_dir": "."} 观察：命令执行完毕。返回码：0 标准输出：Requirement already satisfied: gym in ... Successfully installed ... 思考：安装成功。 行动：execute_shell_command 行动输入：{"command": "python -c \"import gym; print(gym.__version__)\"", "work_dir": "."} 观察：命令执行完毕。返回码：0 标准输出：0.26.2 思考：环境检查通过。 最终答案：步骤1完成。已成功安装并验证gym和stable-baselines3。 步骤 1 结果：步骤1完成。已成功安装并验证gym和stable-baselines3。 === 执行步骤 2 === 描述：编写一个Python脚本，用于定义PPO模型并启动在CartPole-v1环境中的训练。 执行者：executor ...

4.2 关键产出物检查

流程运行后，检查workspace/目录下的内容：

1. plans/：里面应有一个 JSON 文件，记录了规划者生成的完整任务分解。
2. scripts/：执行者智能体生成的 Python 训练脚本应保存在这里，例如train_cartpole_ppo.py。
3. logs/：训练过程的输出日志、评估报告应被保存至此。

你可以查看生成的训练脚本，它应该是一个可运行的、符合stable-baselines3规范的脚本。

4.3 验证训练结果

如果执行者成功运行了训练脚本，你可以在工作区或脚本指定路径找到保存的模型文件（例如ppo_cartpole.zip）。你可以手动编写一个简单的验证脚本，或扩展评估者智能体来自动加载模型并测试其性能。

# manual_test.py - 手动验证训练结果 import gym from stable_baselines3 import PPO env = gym.make("CartPole-v1", render_mode="human") model = PPO.load("workspace/scripts/ppo_cartpole.zip") # 假设模型保存于此 obs, _ = env.reset() for i in range(1000): action, _states = model.predict(obs, deterministic=True) obs, reward, terminated, truncated, info = env.step(action) if terminated or truncated: obs, _ = env.reset() env.close()

如果小车杆子能保持长时间平衡，说明自主训练流程成功产出了一个有效的模型。

5. 常见问题排查与优化

在实际运行中，你可能会遇到各种问题。以下是典型问题的排查路径。

5.1 智能体执行问题排查表

5.2 性能与成本优化

• 模型选型：Gemini 1.5 Flash 是性价比之选，响应快、成本低，适合作为执行者和评估者。对于规划者，可以使用能力更强的 Gemini 1.5 Pro，以获得更可靠、更复杂的规划能力。
• 上下文管理：智能体的“记忆”会消耗上下文窗口。定期总结历史对话，只保留关键信息，或将长上下文存入向量数据库供检索。
• 工具设计：工具应尽可能原子化和可靠。一个经常失败的工具会打乱整个智能体的节奏。为工具添加充分的错误处理和清晰的返回信息。
• 流程短路：并非所有步骤都需要 LLM 思考。对于“检查文件是否存在”、“安装固定依赖”这类确定性任务，可以直接用程序逻辑判断，减少 API 调用。

5.3 安全性强化建议

当前示例中的工具非常强大，但也危险。在生产环境中，必须严格限制：

1. 沙箱化执行：代码和 Shell 命令必须在 Docker 容器或严格隔离的沙箱环境中运行，防止对主机系统的破坏。
2. 命令白名单：实现一个命令过滤器，只允许执行预定义的安全命令列表中的指令。
3. 权限最小化：运行智能体系统的进程应具有最低必要的文件系统权限和网络访问权限。
4. 人工审核环节：在关键步骤（如运行来自网络的脚本、删除文件）前，引入人工确认或审批流程。

6. 扩展方向与最佳实践

这个基础框架可以沿多个方向扩展，以适应更复杂的生产需求。

6.1 系统扩展方向

• 更多角色：引入代码审查者、数据管理者、部署专家等角色，让分工更细。
• 动态工具注册：让智能体在运行时发现并注册新工具，提高系统灵活性。
• 长期记忆与知识库：将成功的计划、解决的 bug、最佳实践存入向量数据库，供后续任务参考，实现“经验”积累。
• 多模态能力：集成 Gemini 的视觉能力，让智能体能分析训练过程中的损失曲线图、模拟器画面等，做出更准确的评估。
• 工作流持久化：将整个协作状态（计划、步骤结果、工作区文件）持久化到数据库，支持暂停、恢复和回滚。

6.2 工程化最佳实践

1. 日志与可观测性：为每个智能体的每次思考、行动和结果记录结构化日志。这不仅是排查问题的关键，也是分析和优化智能体行为的数据基础。
2. 配置外置：将模型参数、温度、提示词模板、工具列表等全部抽取到配置文件（如config.yaml）中，便于在不同环境（开发、测试、生产）切换和进行 A/B 测试。
3. 测试策略：
   • 单元测试：单独测试每个工具函数的正确性。
   • 集成测试：测试两个智能体之间的协作（如规划者-执行者）。
   • 端到端测试：用一组固定的、已知答案的高层目标来运行整个系统，验证最终产出是否符合预期。
4. 优雅降级：当某个智能体（如评估者）多次失败时，系统应能降级到简单规则（如“只要脚本运行无报错即视为成功”）或触发人工干预，而不是完全崩溃。
5. 成本监控与预警：建立 API 调用成本和 token 消耗的实时监控，在超出预算时发出警报或自动暂停非关键任务。

通过以上步骤，我们构建了一个由 Gemini 3.7 Flash 驱动的、具备初步自主协作能力的智能体系统原型。它展示了如何将大语言模型的推理能力与程序化的工具调用相结合，来自动化一个复杂的、多步骤的 AI 开发任务。虽然距离完全自主的 AI 工程师还有很长的路，但此框架为构建自动化程度更高的 AI 研发助手和流水线提供了坚实的技术起点和可扩展的架构。在实际应用中，请务必牢记安全边界，并从小范围、低风险的场景开始逐步验证和迭代。