在实际 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 自主训练机器人模型意味着什么“自主训练”在此处指代一个高度自动化的流程系统在给定高层目标如“训练一个能在模拟环境中移动的机器人模型”和必要资源如模拟器、计算环境后能够自行完成以下循环分析需求理解任务目标和技术约束如使用 PyTorch 还是 TensorFlow是否需要特定传感器数据。生成计划创建数据准备、模型选择、训练、验证的步骤。执行代码编写或调用现有的数据预处理、模型定义和训练脚本。监控与调整运行训练监控损失函数和评估指标根据情况调整超参数或重新规划。输出结果保存训练好的模型并生成训练报告。这个过程的核心挑战在于如何将非结构化的高级目标转化为一系列可被可靠执行的结构化操作代码、命令并处理执行过程中出现的各种异常如库缺失、脚本错误、训练发散。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 jupyterlab2.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 密钥访问 Google AI Studio (https://aistudio.google.com/)。登录你的 Google 账号。在界面中你应该可以找到创建 API 密钥的选项。生成一个密钥并复制。在项目根目录创建一个名为.env的文件将密钥存入# .env 文件内容 GOOGLE_API_KEY你的_实际_API_密钥_在这里重要安全提示务必在.gitignore文件中加入.env切勿将 API 密钥提交到版本控制系统。2.4 辅助工具准备模拟训练环境为了演示“训练机器人模型”我们需要一个简单的训练任务和环境。这里我们选择gymOpenAI Gym和stable-baselines3这个强化学习库它们能让我们快速搭建一个机器人或智能体在模拟环境中学习的案例。这纯粹为了提供具体的、可执行的训练目标。# 安装强化学习模拟环境和库 pip install gym0.26.2 # 注意版本兼容性 pip install stable-baselines3[extra] pip install pyglet1.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(default30, 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_outputTrue, textTrue, timeouttimeout ) 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, shellTrue, capture_outputTrue, textTrue, timeout60 ) 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( modelgemini-1.5-flash, # 或 gemini-1.5-pro根据可用性调整 temperature0.1, # 低温度保证计划更确定、可重复 google_api_keyos.getenv(GOOGLE_API_KEY) ) # 规划者可以使用的工具主要是信息获取和简单验证 planning_tools [ Tool( nameInspect Directory, funclambda 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( llmllm, toolsplanning_tools, promptPLANNER_PROMPT ) planner_executor AgentExecutor( agentplanner, toolsplanning_tools, handle_parsing_errorsTrue, verboseTrue # 调试时打开生产环境关闭 ) 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 fworkspace/plans/plan_{hash(user_goal)}.json os.makedirs(os.path.dirname(plan_file), exist_okTrue) with open(plan_file, w) as f: json.dump(plan_json, f, indent2) 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/目录下的内容plans/里面应有一个 JSON 文件记录了规划者生成的完整任务分解。scripts/执行者智能体生成的 Python 训练脚本应保存在这里例如train_cartpole_ppo.py。logs/训练过程的输出日志、评估报告应被保存至此。你可以查看生成的训练脚本它应该是一个可运行的、符合stable-baselines3规范的脚本。4.3 验证训练结果如果执行者成功运行了训练脚本你可以在工作区或脚本指定路径找到保存的模型文件例如ppo_cartpole.zip。你可以手动编写一个简单的验证脚本或扩展评估者智能体来自动加载模型并测试其性能。# manual_test.py - 手动验证训练结果 import gym from stable_baselines3 import PPO env gym.make(CartPole-v1, render_modehuman) model PPO.load(workspace/scripts/ppo_cartpole.zip) # 假设模型保存于此 obs, _ env.reset() for i in range(1000): action, _states model.predict(obs, deterministicTrue) obs, reward, terminated, truncated, info env.step(action) if terminated or truncated: obs, _ env.reset() env.close()如果小车杆子能保持长时间平衡说明自主训练流程成功产出了一个有效的模型。5. 常见问题排查与优化在实际运行中你可能会遇到各种问题。以下是典型问题的排查路径。5.1 智能体执行问题排查表问题现象可能原因检查点解决方案规划者输出非 JSON提示词约束力不够模型温度过高。查看main.py中planning_agent.invoke的原始输出。1. 强化提示词中“只输出 JSON”的指令。2. 将模型temperature参数降至 0.1 或 0。3. 在代码中添加后处理尝试提取 JSON 部分。执行者卡在“思考”循环工具描述不清任务描述模糊导致模型无法选择工具。观察 LangChain 的verboseTrue输出看模型在“思考”什么。1. 为每个工具编写更精确、无歧义的description。2. 在给执行者的任务描述中明确指出应使用的工具或操作类型如“请使用 Shell 命令工具安装”。Shell 命令执行失败权限不足命令语法错误工作目录不存在。查看ShellCommandTool返回的错误信息result.stderr。1. 在工具内部添加更详细的错误捕获和日志。2. 在执行前让智能体先检查工作目录。3. 避免在智能体中使用需要交互式输入的命令。生成的代码有语法错误模型幻觉上下文不完整。让执行者先运行语法检查命令如python -m py_compile script.py。1. 引入一个“代码审查”步骤在运行前先用简单检查验证。2. 在提示词中要求模型输出“完整、可直接运行”的代码片段。依赖步骤死锁计划中步骤依赖关系出现循环。检查规划者生成的 JSON 中steps的dependencies字段。1. 在规划者提示词中强调“步骤依赖必须是单向的不能循环”。2. 在主控循环run方法中添加循环依赖检测并触发重新规划。API 调用超限或报错Gemini API 调用频率或额度超限网络问题。查看google.generativeai库抛出的异常信息。1. 在代码中添加重试机制和指数退避。2. 监控 API 使用量。3. 对于长流程考虑在步骤间增加延迟。5.2 性能与成本优化模型选型Gemini 1.5 Flash 是性价比之选响应快、成本低适合作为执行者和评估者。对于规划者可以使用能力更强的 Gemini 1.5 Pro以获得更可靠、更复杂的规划能力。上下文管理智能体的“记忆”会消耗上下文窗口。定期总结历史对话只保留关键信息或将长上下文存入向量数据库供检索。工具设计工具应尽可能原子化和可靠。一个经常失败的工具会打乱整个智能体的节奏。为工具添加充分的错误处理和清晰的返回信息。流程短路并非所有步骤都需要 LLM 思考。对于“检查文件是否存在”、“安装固定依赖”这类确定性任务可以直接用程序逻辑判断减少 API 调用。5.3 安全性强化建议当前示例中的工具非常强大但也危险。在生产环境中必须严格限制沙箱化执行代码和 Shell 命令必须在 Docker 容器或严格隔离的沙箱环境中运行防止对主机系统的破坏。命令白名单实现一个命令过滤器只允许执行预定义的安全命令列表中的指令。权限最小化运行智能体系统的进程应具有最低必要的文件系统权限和网络访问权限。人工审核环节在关键步骤如运行来自网络的脚本、删除文件前引入人工确认或审批流程。6. 扩展方向与最佳实践这个基础框架可以沿多个方向扩展以适应更复杂的生产需求。6.1 系统扩展方向更多角色引入代码审查者、数据管理者、部署专家等角色让分工更细。动态工具注册让智能体在运行时发现并注册新工具提高系统灵活性。长期记忆与知识库将成功的计划、解决的 bug、最佳实践存入向量数据库供后续任务参考实现“经验”积累。多模态能力集成 Gemini 的视觉能力让智能体能分析训练过程中的损失曲线图、模拟器画面等做出更准确的评估。工作流持久化将整个协作状态计划、步骤结果、工作区文件持久化到数据库支持暂停、恢复和回滚。6.2 工程化最佳实践日志与可观测性为每个智能体的每次思考、行动和结果记录结构化日志。这不仅是排查问题的关键也是分析和优化智能体行为的数据基础。配置外置将模型参数、温度、提示词模板、工具列表等全部抽取到配置文件如config.yaml中便于在不同环境开发、测试、生产切换和进行 A/B 测试。测试策略单元测试单独测试每个工具函数的正确性。集成测试测试两个智能体之间的协作如规划者-执行者。端到端测试用一组固定的、已知答案的高层目标来运行整个系统验证最终产出是否符合预期。优雅降级当某个智能体如评估者多次失败时系统应能降级到简单规则如“只要脚本运行无报错即视为成功”或触发人工干预而不是完全崩溃。成本监控与预警建立 API 调用成本和 token 消耗的实时监控在超出预算时发出警报或自动暂停非关键任务。通过以上步骤我们构建了一个由 Gemini 3.7 Flash 驱动的、具备初步自主协作能力的智能体系统原型。它展示了如何将大语言模型的推理能力与程序化的工具调用相结合来自动化一个复杂的、多步骤的 AI 开发任务。虽然距离完全自主的 AI 工程师还有很长的路但此框架为构建自动化程度更高的 AI 研发助手和流水线提供了坚实的技术起点和可扩展的架构。在实际应用中请务必牢记安全边界并从小范围、低风险的场景开始逐步验证和迭代。 
长河君 · 2026/8/17 5:11:32 · 阅读 1.2k 基于Gemini与智能体协作的AI模型自主训练系统构建
在实际 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 自主训练机器人模型意味着什么“自主训练”在此处指代一个高度自动化的流程系统在给定高层目标如“训练一个能在模拟环境中移动的机器人模型”和必要资源如模拟器、计算环境后能够自行完成以下循环分析需求理解任务目标和技术约束如使用 PyTorch 还是 TensorFlow是否需要特定传感器数据。生成计划创建数据准备、模型选择、训练、验证的步骤。执行代码编写或调用现有的数据预处理、模型定义和训练脚本。监控与调整运行训练监控损失函数和评估指标根据情况调整超参数或重新规划。输出结果保存训练好的模型并生成训练报告。这个过程的核心挑战在于如何将非结构化的高级目标转化为一系列可被可靠执行的结构化操作代码、命令并处理执行过程中出现的各种异常如库缺失、脚本错误、训练发散。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 jupyterlab2.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 密钥访问 Google AI Studio (https://aistudio.google.com/)。登录你的 Google 账号。在界面中你应该可以找到创建 API 密钥的选项。生成一个密钥并复制。在项目根目录创建一个名为.env的文件将密钥存入# .env 文件内容 GOOGLE_API_KEY你的_实际_API_密钥_在这里重要安全提示务必在.gitignore文件中加入.env切勿将 API 密钥提交到版本控制系统。2.4 辅助工具准备模拟训练环境为了演示“训练机器人模型”我们需要一个简单的训练任务和环境。这里我们选择gymOpenAI Gym和stable-baselines3这个强化学习库它们能让我们快速搭建一个机器人或智能体在模拟环境中学习的案例。这纯粹为了提供具体的、可执行的训练目标。# 安装强化学习模拟环境和库 pip install gym0.26.2 # 注意版本兼容性 pip install stable-baselines3[extra] pip install pyglet1.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(default30, 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_outputTrue, textTrue, timeouttimeout ) 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, shellTrue, capture_outputTrue, textTrue, timeout60 ) 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( modelgemini-1.5-flash, # 或 gemini-1.5-pro根据可用性调整 temperature0.1, # 低温度保证计划更确定、可重复 google_api_keyos.getenv(GOOGLE_API_KEY) ) # 规划者可以使用的工具主要是信息获取和简单验证 planning_tools [ Tool( nameInspect Directory, funclambda 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( llmllm, toolsplanning_tools, promptPLANNER_PROMPT ) planner_executor AgentExecutor( agentplanner, toolsplanning_tools, handle_parsing_errorsTrue, verboseTrue # 调试时打开生产环境关闭 ) 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 fworkspace/plans/plan_{hash(user_goal)}.json os.makedirs(os.path.dirname(plan_file), exist_okTrue) with open(plan_file, w) as f: json.dump(plan_json, f, indent2) 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/目录下的内容plans/里面应有一个 JSON 文件记录了规划者生成的完整任务分解。scripts/执行者智能体生成的 Python 训练脚本应保存在这里例如train_cartpole_ppo.py。logs/训练过程的输出日志、评估报告应被保存至此。你可以查看生成的训练脚本它应该是一个可运行的、符合stable-baselines3规范的脚本。4.3 验证训练结果如果执行者成功运行了训练脚本你可以在工作区或脚本指定路径找到保存的模型文件例如ppo_cartpole.zip。你可以手动编写一个简单的验证脚本或扩展评估者智能体来自动加载模型并测试其性能。# manual_test.py - 手动验证训练结果 import gym from stable_baselines3 import PPO env gym.make(CartPole-v1, render_modehuman) model PPO.load(workspace/scripts/ppo_cartpole.zip) # 假设模型保存于此 obs, _ env.reset() for i in range(1000): action, _states model.predict(obs, deterministicTrue) obs, reward, terminated, truncated, info env.step(action) if terminated or truncated: obs, _ env.reset() env.close()如果小车杆子能保持长时间平衡说明自主训练流程成功产出了一个有效的模型。5. 常见问题排查与优化在实际运行中你可能会遇到各种问题。以下是典型问题的排查路径。5.1 智能体执行问题排查表问题现象可能原因检查点解决方案规划者输出非 JSON提示词约束力不够模型温度过高。查看main.py中planning_agent.invoke的原始输出。1. 强化提示词中“只输出 JSON”的指令。2. 将模型temperature参数降至 0.1 或 0。3. 在代码中添加后处理尝试提取 JSON 部分。执行者卡在“思考”循环工具描述不清任务描述模糊导致模型无法选择工具。观察 LangChain 的verboseTrue输出看模型在“思考”什么。1. 为每个工具编写更精确、无歧义的description。2. 在给执行者的任务描述中明确指出应使用的工具或操作类型如“请使用 Shell 命令工具安装”。Shell 命令执行失败权限不足命令语法错误工作目录不存在。查看ShellCommandTool返回的错误信息result.stderr。1. 在工具内部添加更详细的错误捕获和日志。2. 在执行前让智能体先检查工作目录。3. 避免在智能体中使用需要交互式输入的命令。生成的代码有语法错误模型幻觉上下文不完整。让执行者先运行语法检查命令如python -m py_compile script.py。1. 引入一个“代码审查”步骤在运行前先用简单检查验证。2. 在提示词中要求模型输出“完整、可直接运行”的代码片段。依赖步骤死锁计划中步骤依赖关系出现循环。检查规划者生成的 JSON 中steps的dependencies字段。1. 在规划者提示词中强调“步骤依赖必须是单向的不能循环”。2. 在主控循环run方法中添加循环依赖检测并触发重新规划。API 调用超限或报错Gemini API 调用频率或额度超限网络问题。查看google.generativeai库抛出的异常信息。1. 在代码中添加重试机制和指数退避。2. 监控 API 使用量。3. 对于长流程考虑在步骤间增加延迟。5.2 性能与成本优化模型选型Gemini 1.5 Flash 是性价比之选响应快、成本低适合作为执行者和评估者。对于规划者可以使用能力更强的 Gemini 1.5 Pro以获得更可靠、更复杂的规划能力。上下文管理智能体的“记忆”会消耗上下文窗口。定期总结历史对话只保留关键信息或将长上下文存入向量数据库供检索。工具设计工具应尽可能原子化和可靠。一个经常失败的工具会打乱整个智能体的节奏。为工具添加充分的错误处理和清晰的返回信息。流程短路并非所有步骤都需要 LLM 思考。对于“检查文件是否存在”、“安装固定依赖”这类确定性任务可以直接用程序逻辑判断减少 API 调用。5.3 安全性强化建议当前示例中的工具非常强大但也危险。在生产环境中必须严格限制沙箱化执行代码和 Shell 命令必须在 Docker 容器或严格隔离的沙箱环境中运行防止对主机系统的破坏。命令白名单实现一个命令过滤器只允许执行预定义的安全命令列表中的指令。权限最小化运行智能体系统的进程应具有最低必要的文件系统权限和网络访问权限。人工审核环节在关键步骤如运行来自网络的脚本、删除文件前引入人工确认或审批流程。6. 扩展方向与最佳实践这个基础框架可以沿多个方向扩展以适应更复杂的生产需求。6.1 系统扩展方向更多角色引入代码审查者、数据管理者、部署专家等角色让分工更细。动态工具注册让智能体在运行时发现并注册新工具提高系统灵活性。长期记忆与知识库将成功的计划、解决的 bug、最佳实践存入向量数据库供后续任务参考实现“经验”积累。多模态能力集成 Gemini 的视觉能力让智能体能分析训练过程中的损失曲线图、模拟器画面等做出更准确的评估。工作流持久化将整个协作状态计划、步骤结果、工作区文件持久化到数据库支持暂停、恢复和回滚。6.2 工程化最佳实践日志与可观测性为每个智能体的每次思考、行动和结果记录结构化日志。这不仅是排查问题的关键也是分析和优化智能体行为的数据基础。配置外置将模型参数、温度、提示词模板、工具列表等全部抽取到配置文件如config.yaml中便于在不同环境开发、测试、生产切换和进行 A/B 测试。测试策略单元测试单独测试每个工具函数的正确性。集成测试测试两个智能体之间的协作如规划者-执行者。端到端测试用一组固定的、已知答案的高层目标来运行整个系统验证最终产出是否符合预期。优雅降级当某个智能体如评估者多次失败时系统应能降级到简单规则如“只要脚本运行无报错即视为成功”或触发人工干预而不是完全崩溃。成本监控与预警建立 API 调用成本和 token 消耗的实时监控在超出预算时发出警报或自动暂停非关键任务。通过以上步骤我们构建了一个由 Gemini 3.7 Flash 驱动的、具备初步自主协作能力的智能体系统原型。它展示了如何将大语言模型的推理能力与程序化的工具调用相结合来自动化一个复杂的、多步骤的 AI 开发任务。虽然距离完全自主的 AI 工程师还有很长的路但此框架为构建自动化程度更高的 AI 研发助手和流水线提供了坚实的技术起点和可扩展的架构。在实际应用中请务必牢记安全边界并从小范围、低风险的场景开始逐步验证和迭代。