最近 AI 圈的热度几乎都集中在 Agent 上各家产品都在往“自动干活”的方向卷。很多人觉得构建一个多智能体系统是件很重的事又要设计框架又要编排流程还要管理工具调用。但实际上如果你只是想先跑通一个能并行处理需求、自动拆解任务的 AI 代理团队工具链已经比想象中成熟很多。本文就基于 Grok Bot 的能力带大家从零组建一支最小可用的 AI 代理团队把“主管、策划、执行、质检”这几个角色跑起来整个流程尽量控制在 20 分钟左右。这套思路同样适用于 AutoGPT、Claude Agent 或其他兼容 OpenAI Function Calling 的模型核心设计可以平滑迁移。文章适合三类读者对 AI Agent 概念有了解但还没完整搭过团队系统的开发者。想基于 Grok 模型能力快速验证多角色协作业态的算法工程师。需要在项目里引入 AI 自动化流程但不想一开始就引入重型框架的实战派。下面直接从设计思路开始拆解。1. AI 代理团队到底是什么1.1 从一个问题开始理解我们平时用 AI 对话时典型流程是这样的用户提问 → 模型直接给出回答这种模式适合一次性问答但遇到复杂任务时就暴露问题。比如让 AI“策划一次新品发布活动”它可能只会给出一篇文章而不会自动想到我先分析目标用户再设计活动主题再拆解渠道投放再生成文案最后检查内容是否合规。AI 代理团队就是解决这个问题的。它模拟的是现实公司里的协作方式用户提出目标 → 主管代理拆解任务 → 执行代理各自领取子任务 → 质检代理审核结果 → 汇总反馈每个代理都有明确的角色定位和职责边界它们通过消息机制协作而不是一个模型从头到尾包办。1.2 为什么需要“团队”而不是单 Agent单个 Agent 也能完成任务但团队模式的优势在于职责隔离每个代理只关注自己的领域提示词更聚焦输出质量更稳定。并行效率多个子任务可以并行执行而不是串行等待。可观测性每个环节都有独立记录方便排查问题。扩展灵活后续接入搜索、数据库、邮件等工具时只需要在对应代理上扩展。1.3 Grok Bot 在团队中的定位Grok Bot 作为底层大模型承担的是“大脑”功能。你可以把它理解为一个具备工具调用能力的模型服务团队里的每个代理都可以通过 API 调用它来完成自己的子任务。在本文的架构中我们不需要复杂的外部框架而是用 Python 写一个轻量的调度器负责定义代理角色。维护任务队列。分发任务并收集结果。将最终结果返回给用户。整个系统只有三个核心文件非常适合作为学习 AI 代理架构的入门项目。2. 环境准备与版本说明2.1 开发环境清单开始之前请确认你的本地环境满足以下条件项目说明操作系统Windows 10/11、macOS 或 Linux 均可Python 版本3.10 或更高版本推荐 3.11包管理工具pip 或 poetryGrok API Key需要注册对应平台账号并获取 API Key开发工具VS Code 或任意 Python IDE版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。如果你使用的是其他模型 API只要接口兼容 OpenAI 格式代码基本不需要大改。2.2 安装依赖创建项目目录并初始化虚拟环境mkdir ai-agent-team cd ai-agent-team python -m venv venv source venv/bin/activate # Windows 平台执行 venv\Scripts\activate安装项目依赖pip install openai python-dotenv这里使用openai官方 Python 包是因为目前多数模型服务包括 Grok、GPT、DeepSeek 等都兼容 OpenAI 的 Chat Completions 接口格式。python-dotenv用于管理环境变量避免把密钥硬编码在代码里。2.3 项目目录规划我们采用一个清晰的小型项目结构ai-agent-team/ ├── .env # 存放 API Key 等环境变量 ├── requirements.txt # 项目依赖 ├── agent.py # 代理角色与消息定义 ├── team.py # 团队调度器 └── main.py # 入口文件用于演示完整流程后面每一份代码都会标清楚文件路径方便对照创建。3. 核心概念与架构拆解3.1 任务分解主管代理的工作方式主管代理其实就是带特定系统提示词的模型调用。它接收用户的原始需求然后调用“任务分解函数”把大任务拆成几个可执行的子任务。这里需要注意代理团队里的“角色”并不是独立运行的模型实例而是同一模型在不同提示词上下文中的不同行为模式。理解这一点很重要它决定了团队系统的开销可以非常小。主管代理的提示词核心逻辑可以概括为你是一个项目主管负责将用户的复杂需求拆解为多个可并行执行的子任务。 每个子任务需要有明确的执行说明和预期产出。 请根据用户的输入输出任务列表。对应到代码层面就是让模型输出结构化数据然后由程序解析并生成任务对象。3.2 消息传递代理间的协作机制代理团队中消息传递是最容易设计过度的地方。对于轻量级团队完全不需要引入消息中间件直接用 Python 的字典和列表就能完成消息流转。协作机制如下主管代理生成任务列表。调度器将每个任务包装成一条消息放入执行代理的任务队列。执行代理从队列中取出任务调用模型把结果放入结果池。质检代理从结果池中取结果对每个结果进行评分和反馈。调度器收集所有质检结果生成最终汇总。用一张流程图来表示就是这样注意这是静态示意不是 Mermaid[用户输入] ↓ [主管代理] → 任务列表 ↓ [调度器] → 任务队列 ↓ [执行代理A] [执行代理B] [执行代理C] ↓ ↓ ↓ [结果池] ← ← ← ← ← ↓ [质检代理] → 审核结果 ↓ [汇总输出]3.3 工具调用让代理真正“做事”Grok Bot 支持工具调用能力。所谓工具调用就是模型在生成回答时如果判断需要外部能力会输出一个结构化的调用请求由代码来真正执行这个调用再把调用结果返回给模型继续生成。在代理团队场景中工具函数可以是任意 Python 函数比如搜索网页。查询数据库。调用内部 API。执行本地脚本。读取文件内容。本文为了让流程精简只演示一个模拟搜索工具。但代码结构上保留了通用扩展点后续接真实工具时只需新增函数并注册一下即可。4. 起步实战单代理的最简调用在搭建团队之前先验证最基本的链路通过 Grok API 完成一次带工具调用的对话。这一步能排除大部分环境问题。4.1 配置环境变量在项目根目录创建.env文件# .env GROK_API_KEYyour_api_key_here GROK_BASE_URLhttps://api.x.ai/v1 GROK_MODELgrok-2-latest说明GROK_API_KEY替换为你自己的密钥。GROK_BASE_URL是接口地址如果使用的是兼容 OpenAI 的网关就填写相应地址。GROK_MODEL是模型名称请以平台实际提供的为准。4.2 创建单代理客户端创建agent.py先写一个通用的模型调用函数# agent.py import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(GROK_API_KEY), base_urlos.getenv(GROK_BASE_URL), ) MODEL_NAME os.getenv(GROK_MODEL, grok-2-latest) def call_llm(messages, toolsNone, temperature0.7): 通用的模型调用入口。 Args: messages: 对话消息列表 tools: 可选的工具列表 temperature: 采样温度 Returns: 模型回复的 choices[0].message params { model: MODEL_NAME, messages: messages, temperature: temperature, } if tools: params[tools] tools response client.chat.completions.create(**params) return response.choices[0].message这段代码是整个团队系统的基础层所有代理都会复用这个函数。4.3 模拟一个搜索工具为了演示工具调用流程我们在agent.py里添加一个模拟搜索函数# agent.py def search_web(query: str) - str: 模拟搜索工具根据关键词返回预设信息。 实际项目中可以替换为真实的搜索引擎 API 调用。 search_result { AI Agent: AI Agent 是能够自主感知环境并采取行动达成目标的人工智能程序。, Grok Bot: Grok Bot 是支持工具调用与多轮对话的 AI 模型服务。, 自动化测试: 自动化测试可以提升回归效率减少人工重复操作。, } for key, value in search_result.items(): if key.lower() in query.lower(): return value return f未找到与 {query} 相关的信息。 tools [ { type: function, function: { name: search_web, description: 搜索互联网信息, parameters: { type: object, properties: { query: { type: string, description: 搜索关键词 } }, required: [query] } } } ]工具函数必须返回一个字符串这样模型才能把工具结果和上下文结合继续生成回答。4.4 主流程调用创建临时文件test_single.py来做验证# test_single.py from agent import call_llm, tools, search_web messages [ {role: system, content: 你是一个信息助手可以使用搜索工具获取最新信息。}, {role: user, content: 请介绍一下 Grok Bot 是什么} ] # 第一轮调用让模型决定是否调用工具 message call_llm(messages, toolstools) if message.tool_calls: # 解析工具调用请求 tool_call message.tool_calls[0] function_name tool_call.function.name function_args eval(tool_call.function.arguments) print(f模型选择调用工具: {function_name}) print(f工具参数: {function_args}) # 执行工具调用 if function_name search_web: tool_result search_web(function_args[query]) # 把工具结果追加到消息列表 messages.append({ role: assistant, content: None, tool_calls: [ { id: tool_call.id, type: function, function: { name: function_name, arguments: tool_call.function.arguments } } ] }) messages.append({ role: tool, tool_call_id: tool_call.id, content: tool_result }) # 第二轮调用让模型基于工具结果生成最终回答 final_message call_llm(messages) print(\n最终回答:) print(final_message.content) else: print(message.content)运行测试python test_single.py预期输出大致如下模型选择调用工具: search_web 工具参数: {query: Grok Bot} 最终回答: Grok Bot 是一个支持工具调用与多轮对话的 AI 模型服务能够通过外部工具获取实时信息并完成任务。走到这一步说明单代理链路已经打通。接下来就可以在此基础上构建团队系统了。5. 完整实战构建 AI 代理团队5.1 定义代理基类我们继续在agent.py中完善代码添加代理基类。# agent.py class BaseAgent: 代理基类所有角色代理都继承自此类。 def __init__(self, name, system_prompt, temperature0.7): self.name name self.system_prompt system_prompt self.temperature temperature self.messages [ {role: system, content: self.system_prompt} ] def run(self, user_input): 执行一次对话任务返回模型回复内容。 self.messages.append({role: user, content: user_input}) response call_llm(self.messages, temperatureself.temperature) content response.content self.messages.append({role: assistant, content: content}) return content def reset(self): 重置对话上下文方便复用代理实例。 self.messages [ {role: system, content: self.system_prompt} ]这个基类的作用是统一代理的对话生命周期。每个代理维护自己的消息列表保证上下文隔离。5.2 创建主管代理主管代理需要输出结构化的任务列表因此我们有两种实现思路让模型输出 JSON然后解析。让模型调用create_task_list工具由工具函数返回任务列表。第二种方式更接近真实的 Agent 工程实践因为模型在工具调用模式下的输出更加规范。我们采用第二种方式。继续在agent.py中添加# agent.py import json class CoordinatorAgent(BaseAgent): 主管代理负责拆解任务。 def __init__(self): system_prompt ( 你是一个严谨的项目主管。你的职责是分析用户的目标 将复杂需求拆解为一系列明确、可执行的子任务。 每个子任务必须包含: 任务名称、任务描述、负责角色。 注意子任务之间尽量相互独立方便并行执行。 ) super().__init__(namecoordinator, system_promptsystem_prompt) self.tools [ { type: function, function: { name: create_task_list, description: 创建子任务列表用于将复杂需求拆分为多个可执行任务, parameters: { type: object, properties: { tasks: { type: array, items: { type: object, properties: { name: {type: string, description: 任务名称}, description: {type: string, description: 任务描述}, role: {type: string, description: 执行该任务的角色名称} }, required: [name, description, role] } } }, required: [tasks] } } } ] def decompose(self, user_goal): 拆解用户目标返回任务列表。 user_msg {role: user, content: f用户目标: {user_goal}} messages self.messages [user_msg] response call_llm(messages, toolsself.tools, temperature0.3) # 优先处理工具调用 if response.tool_calls: for call in response.tool_calls: if call.function.name create_task_list: args json.loads(call.function.arguments) tasks args[tasks] # 记录到消息上下文保持对话一致性 self.messages.append(user_msg) self.messages.append({ role: assistant, content: None, tool_calls: [ { id: call.id, type: function, function: { name: call.function.name, arguments: call.function.arguments } } ] }) self.messages.append({ role: tool, tool_call_id: call.id, content: json.dumps(tasks, ensure_asciiFalse) }) return tasks # 如果模型没有走工具调用尝试解析 content 里的 JSON content response.content if content: try: data json.loads(content) if tasks in data: return data[tasks] except json.JSONDecodeError: pass return []主管代理是团队的决策中枢这里我们使用较低的温度0.3让输出更稳定、更少随机性。5.3 创建执行代理执行代理负责具体干活。每个执行代理有一个专属角色名对应不同的提示词。# agent.py class WorkerAgent(BaseAgent): 执行代理负责执行具体子任务。 def __init__(self, role_name): role_prompts { 文案策划: ( 你是一名资深文案策划擅长根据任务要求撰写高质量的文案。 输出内容要求逻辑清晰、语言生动、符合受众需求。 ), 数据分析: ( 你是一名数据分析师擅长处理数据、分析趋势、给出结论。 输出内容要求数据感强、结论明确、有可操作性。 ), 技术方案: ( 你是一名技术架构师擅长编写技术方案、评估可行性、给出技术建议。 输出内容要求结构严谨、考虑边界条件。 ), } system_prompt role_prompts.get( role_name, f你是一名专业执行者负责高效完成分配给你的任务。你的角色是{role_name}。 ) super().__init__(namerole_name, system_promptsystem_prompt) def execute(self, task_name, task_description): 执行任务。 user_input f请完成任务【{task_name}】。任务要求{task_description} return self.run(user_input)你可以根据自己的场景扩充角色提示词例如增加“视觉设计”“客服回复”“代码开发”等角色。5.4 创建质检代理质检代理负责审核执行代理的产出保证质量标准。# agent.py class QCAgent(BaseAgent): 质检代理对任务结果进行质量审核。 def __init__(self): system_prompt ( 你是一个严格的质量审核专家。你会收到一个原始任务及其完成结果 你需要从以下维度进行评估\n 1. 相关度结果是否紧扣任务要求。\n 2. 完整性结果是否覆盖任务的每个核心要点。\n 3. 专业性内容是否体现专业水准。\n 4. 可执行性结果是否具备实际操作价值。\n 最终输出JSON格式{score: 0-100, passed: true/false, suggestions: 改进建议} ) super().__init__(nameqc, system_promptsystem_prompt) def review(self, task_name, task_description, result): 审核单个任务结果。 user_input ( f任务名称{task_name}\n f任务要求{task_description}\n f执行结果\n{result}\n f请对执行结果进行质量审核。 ) review_text self.run(user_input) # 尝试从回复中提取 JSON try: # 简单处理找到大括号内容 start review_text.find({) end review_text.rfind(}) 1 if start ! -1 and end start: json_str review_text[start:end] return json.loads(json_str) except json.JSONDecodeError: pass return { score: 0, passed: False, suggestions: 质检结果解析失败请检查模型输出。 }质检环节在真实项目里很重要。它相当于给 AI 团队上了“保险”避免低质量结果直接流向用户。5.5 创建团队调度器接下来把各个代理组合起来形成完整的协作流程。创建team.py# team.py from agent import CoordinatorAgent, WorkerAgent, QCAgent from concurrent.futures import ThreadPoolExecutor class AgentTeam: AI 代理团队调度器。 def __init__(self): self.coordinator CoordinatorAgent() self.workers {} self.qc_agent QCAgent() def _get_worker(self, role): 根据角色名获取或创建执行代理。 if role not in self.workers: self.workers[role] WorkerAgent(role) return self.workers[role] def run_parallel(self, tasks): 并行执行任务列表。 results {} def execute_one(task): worker self._get_worker(task[role]) result worker.execute(task[name], task[description]) return task, result with ThreadPoolExecutor(max_workers4) as executor: futures [executor.submit(execute_one, task) for task in tasks] for future in futures: task, result future.result() results[task[name]] { task: task, result: result } return results def run(self, user_goal): 完整流程拆解 → 并行执行 → 质检 → 汇总。 print( * 50) print(▶ 阶段一任务拆解) print( * 50) tasks self.coordinator.decompose(user_goal) if not tasks: return 任务拆解失败请重试。 print(f拆解出 {len(tasks)} 个子任务) for i, task in enumerate(tasks, 1): print(f {i}. [{task[role]}] {task[name]}) print(\n * 50) print(▶ 阶段二并行执行) print( * 50) results self.run_parallel(tasks) for task_name, item in results.items(): print(f\n ✅ 任务完成{task_name}) print(f 执行角色{item[task][role]}) print(\n * 50) print(▶ 阶段三质量审核) print( * 50) review_results [] for task_name, item in results.items(): task item[task] result item[result] review self.qc_agent.review( task_nametask[name], task_descriptiontask[description], resultresult ) review_results.append({ task_name: task_name, result: result, review: review }) status 通过 if review.get(passed) else 未通过 print(f {task_name}评分 {review.get(score)}{status}) print(\n * 50) print(▶ 阶段四汇总产出) print( * 50) final_output [] for item in review_results: final_output.append({ task_name: item[task_name], result: item[result], score: item[review].get(score), suggestions: item[review].get(suggestions) }) return final_output调度器使用了ThreadPoolExecutor实现简单的并行执行。对于 I/O 密集型的模型调用这种方式已经能获得不错的并行收益。5.6 编写主入口最后创建main.py# main.py from team import AgentTeam def main(): team AgentTeam() user_goal ( 为我们的新产品‘智能笔记助手’制定一份完整的发布方案 包括目标用户分析、核心文案口号、以及技术实现建议。 ) output team.run(user_goal) print(\n\n 最终交付 \n) for item in output: print(f◆ 任务{item[task_name]}) print(f 评分{item[score]}) print(f 建议{item[suggestions]}) print(f 内容预览{item[result][:200]}...) print() if __name__ __main__: main()5.7 运行完整流程执行以下命令python main.py预期会看到类似下面的输出结构 ▶ 阶段一任务拆解 拆解出 3 个子任务 1. [数据分析] 目标用户画像分析 2. [文案策划] 品牌口号与发布文案 3. [技术方案] 核心技术实现建议 ▶ 阶段二并行执行 ✅ 任务完成目标用户画像分析 执行角色数据分析 ✅ 任务完成品牌口号与发布文案 执行角色文案策划 ✅ 任务完成核心技术实现建议 执行角色技术方案 ▶ 阶段三质量审核 目标用户画像分析评分 88通过 品牌口号与发布文案评分 92通过 核心技术实现建议评分 85通过 ▶ 阶段四汇总产出 整个流程跑通后你就拥有了一支最小可用的 AI 代理团队。6. 进阶扩展给代理团队接入真实能力6.1 接入真实搜索工具把模拟的search_web函数替换为真实的搜索 API。通常步骤是申请搜索服务的 API Key。在.env中添加相关配置。在工具函数中通过 HTTP 请求调用搜索接口。将返回结果清洗后返回字符串给模型。示例代码思路# agent.py import requests def search_web(query: str) - str: 真实搜索工具示例需替换为实际API。 api_key os.getenv(SEARCH_API_KEY) url https://api.example.com/search params {q: query, key: api_key} try: resp requests.get(url, paramsparams, timeout10) data resp.json() return data.get(result, ) except Exception as e: return f搜索失败: {str(e)}注意具体接口地址和字段以你使用的服务商文档为准。6.2 支持自定义工具注册为了让团队系统更通用可以给BaseAgent增加一个工具注册机制# agent.py class BaseAgent: def __init__(self, name, system_prompt, temperature0.7): self.name name self.system_prompt system_prompt self.temperature temperature self.tools [] self.tool_functions {} self.messages [ {role: system, content: self.system_prompt} ] def register_tool(self, tool_definition, function): 注册工具定义及其对应的执行函数。 self.tools.append(tool_definition) self.tool_functions[tool_definition[function][name]] function然后在调用时遍历执行工具调用即可。这种设计让代理能力可以灵活扩展。6.3 延长 Agent 的“记忆”能力当前示例中执行代理每次任务都会累计上下文这在多次调用时可能导致消息列表过长。建议在实际项目里增加上下文管理策略固定窗口只保留最近 N 轮消息。摘要压缩当上下文超过阈值时让模型先总结历史再继续对话。会话隔离每个任务使用独立的代理实例任务结束后释放。更稳妥的方案是在任务执行前调用reset()方法确保上下文干净。7. 常见问题与排查思路7.1 常见报错处理问题现象常见原因解决思路API 连接超时网络不稳定或接口地址错误检查GROK_BASE_URL是否正确确认网络能访问对应服务401 UnauthorizedAPI Key 无效或已过期到平台后台重新生成 Key并确认.env已正确加载模型返回内容为空工具调用后未正确拼接工具结果检查 tool_call_id 是否对应以及 tool 消息格式是否正确任务拆解为空模型输出格式不匹配降低温度参数检查提示词是否明确要求输出结构并行执行报线程异常同一代理实例被多个线程共享为每个任务创建独立代理实例或在执行前加锁质检结果 JSON 解析失败模型输出包含额外文本参考示例中的大括号提取逻辑或改用正则解析7.2 排查步骤建议遇到问题时按以下顺序排查先跑通test_single.py确认最基础的模型调用链路正常。检查环境变量是否正确加载可以在agent.py中打印MODEL_NAME做验证。检查消息格式是否符合 OpenAI Chat Completions 规范。检查工具调用的 function arguments 是否能够被正确解析。查看模型返回的原始响应用response.model_dump_json()打印完整结构。8. 最佳实践与工程建议8.1 提示词设计原则代理团队的提示词是整个系统的灵魂。设计时注意角色边界清晰每个代理的 system prompt 只描述自己的职责不要越界。输出格式明确如果希望模型输出 JSON必须在提示词中明确字段含义和格式要求。给示例优于给规则对复杂输出格式在提示词中给一个一次性的示例效果比抽象描述好得多。温度分场景设置拆解任务、质检审核用低温创意文案、头脑风暴用高温。8.2 代码架构建议配置外置所有密钥、模型名称、URL 都放.env不要硬编码。日志记录为每个代理增加日志输出建议使用logging模块方便追踪调用链路。错误重试对模型调用加入重试机制处理偶发的网络超时。单元测试为核心逻辑任务拆解、结果解析、质检解析编写测试用例。预设降级策略如果质检不通过可设置自动重跑或标记人工审核。8.3 生产环境注意事项成本和速率控制给每个代理增加 Token 用量统计设置最大调用次数限制。安全边界涉及外部工具的调用必须做权限校验和输入过滤避免 Agent 产生越权行为。数据隐私敏感数据不要直接放入系统提示词或上下文建议脱敏。可观测性把每次任务的输入、输出、耗时、评分记录到数据库形成可追溯的日志体系。备份方案模型服务可能变更或限流建议保留一个基础模型作为降级方案。8.4 性能优化合理设置并行度不是越大越好要考虑 API 的速率限制。上下文裁剪对于超长结果在送入质检代理前先做摘要节省 Token。结果缓存对相似度极高的请求可以缓存模型输出降低调用成本。异步处理在 Web 场景中建议把团队运行逻辑封装为异步任务避免阻塞接口响应。9. 扩展学习方向如果这篇文章的示例你已经全部跑通下一步可以往下面这些方向深耕引入记忆机制给团队增加持久化存储让代理能记住多轮任务的上下文。接入外部知识库通过 RAG 方式让代理在执行任务时能查询企业内部文档。构建可视化编排界面用 Gradio 或 Streamlit 给团队系统加一个 Web 交互界面方便非技术人员操作。补充人工审批环节在关键节点加入人工确认提升系统安全性。研究模型差异在不同模型服务之间做对比测试找出最适合 Agent 场景的模型配置。AI 代理团队的设计空间非常广阔从一个简单的原型开始逐步加记忆、加工具、加权限控制最后就能演变成一个真正能辅助业务运行的智能体系统。关键不在于用了多复杂的框架而在于你能不能先让最基础的“拆解-执行-质检”闭环跑起来。架构可以慢慢演进但最小闭环越早打通越好。