Grok Bot多代理协作实战:Python实现最小AI代理团队 📅 发布时间:2026/9/3 17:44:36 👁 浏览次数: 最近一段时间只要打开技术社区几乎每隔几天就能看到一篇关于 AI 代理AI Agent的文章。有人把它描述成“给大模型装上手脚”也有人已经用多代理协作处理复杂业务。但真到自己上手时大部分人都会卡在同一个地方概念读了一堆代码不知道从哪里开始写。这里藏着一个认知误区很多人以为多代理的难点在于“让模型变聪明”但在实际工程里模型只是推理引擎真正决定一个代理团队能不能跑起来的是角色分工、消息路由和结果汇合这三件事。模型负责“想”你要负责“组织”。这篇文章的目标是给你一套可以直接抄的最小实现。我们用 Grok Bot 作为推理核心用 Python 写一个包含产品经理、编码员、审查员三个角色的 AI 代理团队从环境准备到跑通完整任务熟练的话 20 分钟左右就能完成。如果你有基本的 Python 基础照着做一遍就能理解多代理协作的核心骨架而不是停留在刷概念的状态。1. 这篇文章真正要解决的问题先聊一个常见的痛点当你拿着一个问题去问 ChatGPT、Grok 这类大模型时它能给你一个不错的答案但回答完之后它不会主动帮你检查也不会把一个复杂需求拆成几步去执行。单次对话模型本质上是一个“单点推理器”它缺乏任务拆解、分工执行、结果验收这样一套流程。多代理团队要解决的问题就是把这个流程补齐。它的核心思想很朴素不要让一个大模型从头到尾包办所有事而是把任务拆给多个具有不同角色定位的 AI 代理让它们像真实团队一样协作。举个例子一个典型的“写代码任务”可以拆成三段流程产品经理代理负责理解需求、拆解任务编码代理负责按任务清单写代码审查代理负责检查代码质量反馈修改意见。这样做的价值在于每个代理只需要专注于自己负责的那一段提示词可以更聚焦输出质量更容易控制。而且审查环节相当于给 AI 生成的内容加了一道校验能明显减少低级错误。这篇文章会围绕这个最小团队展开全部实现。读完你会得到三样东西一套可以运行的 Grok Bot 多代理代码骨架代理之间如何传消息、如何判断流程终点的完整逻辑从单机脚本扩展到工程化多代理系统时需要避开的坑。如果你是刚接触 AI 代理的工具使用者这篇文章可以帮你建立全局认知如果你是有一定经验的开发者直接用第三、四、五章代码即可。2. 基础概念Grok Bot、AI 代理与代理团队2.1 Grok Bot 是什么Grok Bot 是基于 Grok 模型的 AI 代理/对话应用。Grok 系列模型擅长自然语言对话和推理类任务而 Grok Bot 则是把这种能力封装成一个可以被用户或程序调用的 Bot 产品。在实际开发中我们更关心的是它的 API 接入方式。Grok 的 API 与 OpenAI 的 Chat Completions 格式兼容也就是说你不需要引入一个完全陌生的 SDK用 Python 生态里常见的openai库把base_url和 API Key 一换就能实现模型调用。这一点会大大降低集成成本。2.2 AI 代理AI Agent和普通模型调用有什么区别普通模型调用是这样的用户输入 - 模型生成 - 返回结果AI 代理则更像一个“会做决策的执行者”用户输入 - 代理理解任务 - 决定执行步骤 - 调用模型/工具 - 检查结果 - 返回最终结果AI 代理的关键能力是任务编排。它不满足于“回答一个问题”而是把一个大目标拆成子任务再决定按什么顺序、用什么工具去执行这些子任务。2.3 代理团队Multi-Agent和单代理的区别单代理虽然能规划任务但它的规划、执行、检查都在同一个上下文里完成很容易出现两个问题一是上下文过长导致模型遗忘前面的信息二是缺少“外部检查”机制模型自己很难发现自己写错了。代理团队把不同职责放到独立的代理中每个代理有独立的系统提示词和上下文窗口。这样做有四个好处角色边界清晰每个代理只负责一件事提示词不会被其他任务干扰上下文隔离代码生成不会被冗长的需求讨论冲淡质量检查审查代理可以站在另一个角度挑毛病可替换性某个环节的模型选型可以独立调整比如审查代理换成更强的小模型。下表对比了三种使用方式对比维度单次对话模型单代理多代理团队目标回答单次问题完成一段任务链完成复杂协作任务上下文管理单轮/多轮对话单个上下文多代理独立上下文任务拆解无可简单拆解可结构化拆解质量检查无依赖模型自觉有独立审查角色实现成本最低中较高适用场景问答、翻译、摘要文档处理、工具调用代码生成、研究报告、复杂流程2.4 核心设计原则调度器模式多代理协作有多种实现方式最简单也最稳定的是“调度器模式”。它的核心是代理之间不直接互相调用而是由一个调度器Orchestrator统一管理消息流。用户请求 - 调度器 ├── 调用产品经理代理 - 得到任务清单 ├── 调用编码代理 - 得到代码 ├── 调用审查代理 - 得到审查意见 └── 决定是否返工或返回结果调度器最大的好处是流程可控。你可以清晰地看到每一轮调用发生了什么可以在任意环节插入日志、增加重试、控制成本。后面第五、六章的代码就是围绕这个模式实现的。3. 环境准备与前置条件3.1 操作系统与 Python 版本本文示例在 macOS / Linux / Windows 上都可以运行。需要安装 Python 3.10 或以上版本一个小检查命令python --version如果你看到输出是 Python 3.10 以上就可以继续。3.2 获取 Grok API Key要调用 Grok 模型需要到对应平台开通 API 并生成一个 Key。这里有两个提醒API Key 是敏感信息只保存在本地环境变量或.env文件中不要提交到代码仓库不同的模型 ID 对应不同能力和价格具体模型名以你的账号在控制台里实际可用的为准不要照抄网络文章里的旧模型名。我的建议是先在控制台确认能正常发起一次对话再进入下面的步骤这样能避免后面配置代理团队时排查半天发现是 Key 的问题。3.3 安装依赖创建一个工作目录比如grok-agent-team然后在目录下创建虚拟环境并安装依赖mkdir grok-agent-team cd grok-agent-team python -m venv venv # macOS / Linux source venv/bin/activate # Windows PowerShell # .\venv\Scripts\Activate.ps1 pip install openai python-dotenv本文只需要两个依赖openai用来调用兼容 OpenAI 格式的 Grok APIpython-dotenv用来从.env文件加载环境变量。3.4 项目文件结构后面会创建如下文件grok-agent-team/ ├── .env # 存放 API Key 等敏感配置 ├── .gitignore # 忽略 .env 和 venv ├── agent.py # Agent 基础类和角色提示词 ├── orchestrator.py # 多代理调度器 └── main.py # 程序入口4. 核心架构设计三个代理怎么分工4.1 为什么选这三个角色一个最小但完整的代理团队至少需要三个角色产品经理代理PM Agent理解用户需求把模糊请求拆成可执行任务清单编码代理Coder Agent按任务清单写代码审查代理Reviewer Agent检查代码发现问题就反馈给编码代理修改。这三个角色形成了“拆解 - 执行 - 验收”的闭环。虽然不是每类任务都需要代码审查但这个结构覆盖了多代理系统最基本的消息流串行传递、条件判断、循环返工。4.2 消息流设计整个流程如下用户输入需求调度器调用 PM 代理得到结构化任务清单调度器把任务清单传给 Coder 代理得到代码调度器把代码传给 Reviewer 代理得到审查意见如果审查意见明确表示通过流程结束如果审查不通过调度器把审查意见拼进新任务让 Coder 代理修改代码然后再次走审查超过最大迭代轮数后强制结束避免无限循环。这里有个容易被忽略的细节Reviewer 的判断不能只依赖模型的“自觉”调度器必须设定一个终止条件。最简单的方式是规定“审查意见中包含某个通过标记”更稳的方式是让审查代理输出 JSON 结构化结果这个后面会展开。4.3 提示词设计像写岗位说明书每个代理的提示词本质上是岗位说明书。它至少要说明三件事你是谁角色你要输出什么交付物格式你不做什么边界。举个例子PM 代理的提示词如果写成“你是一个产品经理”模型就不知道输出格式和边界可能出现天马行空的结果。但如果加上“输出 3 到 6 条任务、每条包含目标和预期产出、不要写代码”这些约束输出质量会稳定很多。5. 完整代码实现20 分钟跑通最小代理团队下面从配置到运行逐步给出完整代码。先说明代码里的base_url和模型名等参数以官方文档和你的账号实际权限为准示例中默认值仅用于演示。5.1 配置环境变量创建.env文件# 文件路径grok-agent-team/.env GROK_API_KEYyour_grok_api_key_here GROK_BASE_URLhttps://api.x.ai/v1 GROK_MODELgrok-2-latest创建.gitignore文件# 文件路径grok-agent-team/.gitignore .env venv/ __pycache__/这里真正容易踩坑的地方是很多人把 API Key 直接写死在代码里然后不小心提交到了公开仓库导致密钥泄露。.env文件配合.gitignore是成本最低的安全习惯。5.2 Agent 基础类封装模型调用创建agent.py# 文件路径grok-agent-team/agent.py import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() GROK_API_KEY os.getenv(GROK_API_KEY) GROK_BASE_URL os.getenv(GROK_BASE_URL, https://api.x.ai/v1) GROK_MODEL os.getenv(GROK_MODEL, grok-2-latest) if not GROK_API_KEY or GROK_API_KEY your_grok_api_key_here: raise RuntimeError(请先在 .env 文件中配置 GROK_API_KEY) client OpenAI(api_keyGROK_API_KEY, base_urlGROK_BASE_URL) class Agent: 一个基本的 AI 代理角色。 name: 代理名称 system_prompt: 角色提示词 temperature: 采样温度越低越稳定 max_tokens: 单次生成的最大 token 数 def __init__( self, name: str, system_prompt: str, temperature: float 0.7, max_tokens: int 1200, ): self.name name self.system_prompt system_prompt self.temperature temperature self.max_tokens max_tokens def run(self, task: str) - str: response client.chat.completions.create( modelGROK_MODEL, messages[ {role: system, content: self.system_prompt}, {role: user, content: task}, ], temperatureself.temperature, max_tokensself.max_tokens, ) return response.choices[0].message.content这段代码的核心是run方法。每个代理本质上就是“系统提示词 用户任务”的组合把一次模型调用封装成带角色的执行者。temperature参数需要留意拆解任务、审查代码这类偏逻辑的环节建议用低温度0.3 左右生成代码可以稍高一点0.5但不要太高否则输出随机性会变大。5.3 定义三个角色的提示词继续在agent.py中补充角色提示词# 文件路径grok-agent-team/agent.py追加 PM_PROMPT 你是产品经理代理PM Agent。你的任务是把用户需求拆成清晰、可执行的技术任务清单。 要求 1. 输出 3 到 6 条具体任务 2. 每条任务包含目标、输入、预期产出 3. 使用编号列表语言简洁 4. 不要写代码只做任务拆解 5. 如果需求本身有歧义先给出你的假设再继续拆解 CODER_PROMPT 你是编码代理Coder Agent。你负责根据产品经理的任务清单编写可运行的 Python 代码。 要求 1. 输出完整代码包含必要的注释 2. 如果任务不清晰先补充你的假设再写代码 3. 代码默认在当前项目目录下运行 4. 只输出代码和简短说明不要长篇幅解释 REVIEWER_PROMPT 你是审查代理Reviewer Agent。你负责检查编码代理输出的代码。 检查重点 1. 语法是否正确 2. 边界条件是否处理 3. 是否存在安全隐患如路径穿越、命令注入 4. 是否可能存在运行时异常 输出要求 1. 如果确认没有问题第一行必须写通过审查 2. 如果发现问题先列出问题列表再给出修改建议 3. 不要擅自修改代码只输出审查意见提示词的可控性直接决定代理团队的上限。如果只是简单写“你是程序员”模型就会按照自己的默认习惯输出格式不稳定。而这里的提示词已经为调度器逻辑服务Reviewer 的输出要求“第一行必须写‘通过审查’”就是为了让调度器能通过字符串判断是否结束流程。5.4 调度器让三个代理协作起来创建orchestrator.py# 文件路径grok-agent-team/orchestrator.py from agent import ( Agent, CODER_PROMPT, PM_PROMPT, REVIEWER_PROMPT, ) class AgentTeam: 一个极简多代理调度器。 max_rounds: 审查不通过时最多让编码代理修改几轮 def __init__(self, max_rounds: int 2): self.max_rounds max_rounds self.pm Agent(PM, PM_PROMPT, temperature0.3, max_tokens800) self.coder Agent(Coder, CODER_PROMPT, temperature0.5, max_tokens1800) self.reviewer Agent(Reviewer, REVIEWER_PROMPT, temperature0.3, max_tokens1200) def run(self, user_request: str) - dict: # 1. 产品经理拆解任务 task_list self.pm.run(user_request) # 2. 编码代理根据任务清单写代码 code self.coder.run(task_list) # 3. 进入“审查 - 修改 - 再审查”的循环 last_review None for round_idx in range(1, self.max_rounds 1): last_review self.reviewer.run(code) if 通过审查 in last_review: return { status: passed, round: round_idx, task_list: task_list, code: code, review: last_review, } # 审查未通过把审查意见作为新任务交给编码代理修改 code self.coder.run( f审查未通过意见如下\n{last_review}\n 请根据审查意见修改代码输出修改后的完整代码。 ) return { status: max_rounds_exceeded, round: self.max_rounds, task_list: task_list, code: code, review: last_review, }调度器的逻辑重点在循环部分。很多人第一次写多代理时会把所有代理的结果一次性打印出来但没有“返工”机制审查代理的意见根本没有被消费掉这就变成了伪多代理。真正让团队协作运转起来的是“审查不通过 - 把意见拼进新任务 - 重新交给编码代理”这一步。5.5 主程序入口创建main.py# 文件路径grok-agent-team/main.py from orchestrator import AgentTeam def main(): request ( 写一个 Python 脚本读取当前目录下的 data.csv 文件 按第二列数值从大到小排序然后把结果保存到 sorted.csv。 ) team AgentTeam(max_rounds2) result team.run(request) print( 任务拆解 ) print(result[task_list]) print() print( 最终代码 ) print(result[code]) print() print( 审查结论 ) print(result[review]) print() print(f 流程状态: {result[status]} ) if __name__ __main__: main()到这里一个最小的 Grok Bot AI 代理团队就完成了。整体流程是main.py 发起请求AgentTeam 调度器依次调用三个代理最后把任务拆解、代码、审查结论和流程状态一起返回。6. 运行结果与效果验证6.1 运行命令在虚拟环境激活状态下执行python main.py6.2 预期输出输出会分三个区域下面是一个演示性质的示意实际内容由模型生成不会逐字相同 任务拆解 1. 目标读取 data.csv 文件输入本地文件预期产出DataFrame 数据 2. 目标按第二列排序输入DataFrame预期产出排序后的 DataFrame 3. 目标保存为 sorted.csv输入排序后的 DataFrame预期产出新文件 最终代码 import pandas as pd df pd.read_csv(data.csv) df_sorted df.sort_values(bydf.columns[1], ascendingFalse) df_sorted.to_csv(sorted.csv, indexFalse) 审查结论 通过审查 代码逻辑正确但建议补充文件不存在时的异常处理。 流程状态: passed 6.3 如何判断成功判断标准有三个程序没有异常退出输出中的流程状态为passed说明审查代理在规定的轮数内确认通过你可以在当前目录下实际生成sorted.csv文件结构符合预期。如果出现max_rounds_exceeded不代表代码写错了只说明审查代理在最大轮数内没有给出“通过审查”的结论。可以先人工看一下审查意见判断是代码真有问题还是提示词判断条件太严。6.4 失败时第一步看哪里如果运行失败我最建议先看报错的第一行而不是直接改代码。比如如果是401相关问题在 API Key如果是model not found问题在模型名如果是网络超时问题在请求链路或参数配置而不是代理逻辑。7. 常见问题与排查方法下表汇总了几类最常见的运行问题问题现象可能原因排查方式解决方案程序启动直接报错提示未配置 GROK_API_KEY.env文件不存在或占位符没替换检查.env文件内容确认GROK_API_KEY不是your_grok_api_key_here填入真实 API Key 并确认文件位于项目根目录请求返回 401 / invalid api keyAPI Key 无效、过期或复制多了空格打印环境变量长度到控制台重新生成 Key重新复制 Key注意不要带换行符请求返回 model not found模型名与当前账号可用模型不一致到控制台确认可用模型列表修改.env中的GROK_MODEL请求一直超时网络不稳定或max_tokens设置过大先测试最小请求降低max_tokens到 500检查网络分段处理长任务审查代理总是返回“不通过”提示词判断条件过严或代码确实有问题打印审查意见全文人工判断放宽REVIEWER_PROMPT的通过标准修改代码逻辑审查代理总是返回“通过”但代码明显有错模型没有真正执行校验只是“礼貌性通过”在提示词中增加具体的检查清单降低 temperature让审查代理输出 JSON 结构化结论规定必须列出检查项多轮修改后代码越改越差每次修改都把整段历史上下文重复传入模型被带偏查看每轮传给编码代理的任务内容在返工时只传递“最近一次审查意见”和“当前代码”关键信息token 消耗增长很快返工轮数设置过高、上下文重复拼接在调度器里打印每次调用的 token 数设置合理max_rounds优先用更强的小模型做审查而不是反复让大模型返工第七个问题值得多说一句上下文污染是多代理系统最常见的质量杀手。返工时如果把“第一次需求 第一次代码 第二次审查意见 第二次代码”全部塞给编码代理模型会迷失在历史信息里改出一个四不像。正确做法是每次返工只传“当前代码 最新审查意见”让代理把注意力集中在当前问题。8. 最佳实践与工程建议跑通最小实现只代表你理解了消息流真正放到业务里还需要补齐下面几件事。8.1 角色提示词要像岗位说明书提示词的核心不是文采而是边界。好的角色提示词应该明确以下内容角色的目标输入数据长什么样交付物的格式什么情况下需要提出假设哪些事情绝对不做。比如“不写代码”“不要长篇幅解释”“第一行必须写‘通过审查’”这些边界条件都直接服务于调度器的稳定性。建议你把每一个代理的提示词都按照“身份 目标 输入 输出格式 边界”的结构来写。8.2 代理之间优先传递结构化消息字符串传递虽然简单但稳定性不够。更工程化的做法是让代理输出 JSON 结构例如审查代理返回{ passed: false, issues: [缺少文件存在性检查], suggestions: [加入 os.path.exists 判断] }调度器解析这个 JSON用passed字段判断流程是否继续比在字符串里匹配“通过审查”要可靠得多。使用时要注意模型不一定每次都能输出合法 JSON建议在解析失败时降级为字符串判断或者提示模型“如果输出 JSON 不合法将被视为审查不通过”。8.3 给调度器设置最大轮数多代理系统最危险的场景是“死循环”。如果审查代理永远不满意编码代理可以无限次修改token 成本会不可控。代码中的max_rounds参数就是最后一道阀门生产环境建议设置为 1 到 3不要更高。8.4 记录每次调用的时间与 token 数多代理工程化运行后成本分析必须跟上。最轻量的方式是给Agent.run方法加一层统计例如# 伪代码示意思路 def run_with_metrics(self, task: str): start time.time() response client.chat.completions.create( modelGROK_MODEL, messages[...], temperatureself.temperature, max_tokensself.max_tokens, ) latency time.time() - start usage response.usage logger.info( fagent{self.name} latency{latency:.2f}s fprompt_tokens{usage.prompt_tokens} fcompletion_tokens{usage.completion_tokens} ) return response.choices[0].message.content这样你就能清楚地知道每个代理分别花了多少钱、耗了多少时间再决定要不要把某些环节替换成更小的模型。8.5 接入本地模型作为私有子代理有人提到“AI 代理助手加本地模型”这其实是一个很实用的工程思路。当任务涉及内部数据但你又不想把敏感内容发送到外部 API 时可以把“数据预处理”“关键词抽取”“简单分类”这类相对低风险的任务路由给本地模型执行让 Grok Bot 只负责最终决策和复杂推理。如果你本机已经安装了 Ollama可以用下面的方式接入# 文件路径grok-agent-team/local_model.py可选 from openai import OpenAI # Ollama 默认提供 OpenAI 兼容接口 local_client OpenAI( base_urlhttp://localhost:11434/v1, api_keyollama, ) def local_chat(prompt: str) - str: response local_client.chat.completions.create( modelqwen2.5:7b, # 以本机 ollama list 实际模型为准 messages[ {role: user, content: prompt}, ], temperature0.3, ) return response.choices[0].message.content把这部分代码接进调度器后你就有了一条“本地模型 云端模型”的混合链路。本地模型负责成本低、隐私敏感的前置处理Grok Bot 负责推理密集的复杂任务。两条链路通过同一个调度器串起来对外仍然是一个代理团队。8.6 安全边界不要让代理直接操作生产环境AI 代理有一个很大的风险它生成的东西看起来很有道理但不代表真的安全。具体到本文场景审查代理能发现一部分代码问题但不等于它是安全审查工具。如果代理生成的代码要写入文件、执行命令、操作数据库必须在沙箱环境中先验证再人工确认。任何时候遵循最小权限原则给代理的 API Key、数据库账号、服务器权限都只开放任务确实需要的范围。9. 总结与后续学习方向这篇文章的核心其实不是 Grok Bot 本身而是围绕它搭建的多代理协作骨架。我们只用了一个模型 API、三个 Python 文件就实现了一个具备“需求拆解 - 代码生成 - 审查返工”能力的 AI 代理团队。这背后真正值得理解的是调度器模式代理之间不直接互相调用所有消息由调度器统一转发流程可控、可观测、可扩展。如果你跟着代码走了一遍下一步建议按这个顺序深入把代理之间的字符串消息改成 JSON体验结构化消息带来的稳定性提升给调度器加日志和 token 统计把成本可视化接一个工具调用能力比如让编码代理真的在沙箱里执行代码而不是只生成代码尝试用本地模型替换其中一个代理对比质量和成本。多代理团队本质上是一个“可以稳定复用的编排流程”。跑通骨架只是开始真正有价值的部分是你往角色里填充的领域知识和边界约束。多试几组提示词多观察几次失败你对这个系统的掌控感会很快超过那些只看概念的人。建议把这份代码骨架保存好后续做任何多代理项目都可以从它快速起步。