1. 项目背景与“智能体团队”的引入

在之前的几篇笔记里,我们已经把 learn-claude-code 这个项目从零开始,一步步搭建了起来。从最基础的环境配置、模型接入,到实现代码补全、对话交互,再到引入文件操作、上下文管理等高级功能,整个智能体(Agent)的骨架已经相当完整了。它就像一个训练有素的“全栈工程师”,能理解你的意图,并独立完成一系列编码任务。

但不知道你有没有遇到过这样的场景:一个复杂的项目,前端、后端、数据库设计、部署脚本,甚至还有文档撰写,需要多种不同的专业技能。让一个“全栈”智能体去处理,虽然也能做,但就像让一个工程师同时切换多个角色,效率未必最高,而且在某些深度领域(比如复杂的算法优化或者特定的框架配置)可能不够“专精”。这时候,一个自然的想法就出现了:能不能组建一个“团队”?让擅长前端的智能体去写界面,让精通后端的去搭服务,再让一个“测试专家”来写单元测试,最后还有个“文档工程师”来整理说明。这就是 Agent Teams(智能体团队) 概念的核心价值。

ClaudeCode 或类似的 AI 编码助手中,实现“智能体团队”并不是指真的启动多个 AI 模型实例(那成本太高了),而是指设计一套精妙的协作机制。让一个“主智能体”(或称为“协调者”、“管理者”)来分解任务、分配工作、整合结果,而具体的执行则由内部不同的“技能模块”或“子流程”来完成。这本质上是一种 “分而治之” “专业化分工” 的软件工程思想在 AI 智能体架构上的体现。通过这种方式,我们可以显著提升复杂任务的完成质量、可解释性和可控性。

2. 智能体团队的架构设计与核心思想

learn-claude-code 项目中实现智能体团队,我们首先要摒弃“一个提示词走天下”的简单思维。我们需要设计一个清晰的架构,来定义团队成员、协作流程和沟通规范。

2.1 角色定义:你的团队需要哪些成员?

一个高效的软件开发团队通常包括这些角色:产品经理(理解需求)、架构师(设计蓝图)、前端工程师、后端工程师、测试工程师、运维工程师等。在我们的 AI 智能体团队中,我们可以抽象出以下几种核心角色:

  1. 任务分析员 (Task Analyst) :它的职责是解读用户模糊或复杂的需求,将其拆解成具体的、可执行的子任务列表。例如,用户说“帮我创建一个简单的待办事项 Web 应用”,分析员需要输出:① 设计数据库表结构;② 创建后端 RESTful API;③ 实现前端页面组件;④ 编写基础样式;⑤ 提供运行说明。
  2. 架构师/技术选型员 (Architect) :负责为整个项目或特定子任务选择合适的技术栈、框架、库,并设计核心的目录结构和模块划分。它会考虑项目的规模、性能要求、团队熟悉度等因素。
  3. 代码实现员 (Coder) :这是最核心的执行角色,根据分配的具体任务和架构设计,编写实际的代码。我们可以进一步细分,比如 Frontend Coder (擅长 React/Vue/HTML/CSS)、 Backend Coder (擅长 Node.js/Python/Go)、 Database Coder (擅长 SQL/Schema设计)。
  4. 代码审查员 (Code Reviewer) :在代码编写完成后,审查员负责检查代码质量,包括但不限于:语法错误、潜在 bug、代码风格一致性、性能问题、安全性问题等。它提供修改建议,确保代码符合标准。
  5. 测试工程师 (Tester) :负责为编写好的代码(尤其是核心功能)编写单元测试、集成测试用例,并可能执行测试,确保功能的正确性和健壮性。
  6. 文档工程师 (Documenter) :负责生成项目 README、API 文档、代码注释等,让项目易于理解和维护。
  7. 项目协调员 (Coordinator) :这是整个团队的“大脑”或“项目经理”。它接收用户原始需求,调用“任务分析员”进行分解,然后根据任务类型分配给不同的“实现员”,并串行或并行地调度“审查员”、“测试员”和“文档员”的工作,最后整合所有输出,呈现给用户。

在实际的 learn-claude-code 实现中,我们可能不会为每个角色都创建一个独立的 AI 调用(那样 token 消耗巨大)。更可行的策略是: 让一个强大的主模型(如 Claude 3.5 Sonnet 或 GPT-4)来扮演“项目协调员” ,并通过精心设计的系统提示词(System Prompt),让它内部模拟出上述不同角色的思维过程,按步骤工作。或者,我们可以设计一套流程,让主模型在不同阶段,切换使用针对不同角色优化的“提示词模板”和“上下文”。

2.2 协作流程:团队如何运转?

一个典型的智能体团队工作流可以设计如下:

  1. 需求接收与澄清 :用户提出需求。协调员首先与用户进行简短对话,澄清模糊点,明确核心功能和约束条件(如技术栈偏好)。
  2. 任务分解与规划 :协调员内部调用“任务分析员”模式,将需求分解为任务列表(Task List),并评估依赖关系和执行顺序。
  3. 技术设计与分配 :对于每个任务,协调员可能切换到“架构师”模式,给出简要的技术设计。然后,它将具体的编码任务分配给“代码实现员”模式。
  4. 循环执行与审查
    • 实现员生成代码。
    • 协调员切换到“代码审查员”模式,对刚生成的代码进行审查,提出修改意见。
    • 实现员根据意见修改代码。此过程可能迭代多次,直到审查通过。
    • 对于关键模块,协调员可能接着切换到“测试工程师”模式,生成测试用例。
  5. 集成与交付 :所有子任务完成后,协调员负责确保不同模块能整合在一起(例如,检查前端 API 调用与后端接口是否匹配)。最后,“文档工程师”模式生成项目文档。协调员将所有代码、文档整理好,交付给用户。

这个流程的关键在于 “上下文管理” 。每个角色“工作”时,它需要看到完整的项目背景、之前已完成的工作、以及它专属的指令。我们需要在代码中维护一个不断增长的“项目上下文”,里面包含了需求描述、任务列表、已生成的代码文件、审查意见、测试用例等。每次调用模型时,我们都从这个上下文中提取相关信息,并附加当前角色需要执行的特定指令。

2.3 沟通媒介:共享工作区与结构化输出

团队成员不能靠“心灵感应”沟通。我们需要定义一个共享的“工作区”。在 learn-claude-code 中,这可以是一个复杂的数据结构或一组文件。

  • 项目状态对象 :一个在内存中维护的字典或对象,记录 current_task , completed_tasks , code_files (字典,key为文件路径,value为内容), review_notes , test_cases 等。
  • 结构化输出约束 :为了便于程序自动解析和处理每个角色的输出,我们必须要求模型以严格的格式(如 JSON、XML 或特定的 Markdown 标签)进行响应。例如,当“代码实现员”完成任务时,它必须输出:
    {
      "action": "write_code",
      "files": [
        {
          "path": "src/main.js",
          "content": "console.log('Hello World');"
        },
        {
          "path": "package.json",
          "content": "{...}"
        }
      ],
      "explanation": "创建了入口文件并初始化了项目配置。"
    }
    
    同样,“代码审查员”的输出可能是:
    {
      "action": "code_review",
      "file_path": "src/main.js",
      "issues": [
        {"line": 1, "suggestion": "建议使用 const 代替 let。"},
        {"line": 5, "suggestion": "这个函数缺少错误处理。"}
      ],
      "approved": false
    }
    
    这种结构化输出使得我们的程序可以自动更新“项目状态对象”,并决定下一步该调用哪个“角色”。

3. 在 learn-claude-code 中实现基础团队协作

理论说完了,我们来看代码。我们不会重写整个项目,而是在现有基础上进行扩展。假设我们已经有一个能处理对话和简单代码生成的 ClaudeCodeAgent 类。现在,我们要创建一个 AgentTeamCoordinator 类。

3.1 定义角色与提示词模板

首先,我们定义各个角色的系统提示词。这些提示词告诉模型“你现在是谁,你要做什么”。

# 提示词模板定义
ROLE_PROMPTS = {
    "coordinator": """你是一个AI软件开发团队的项目协调员。你的工作是管理整个项目流程。
1. 理解用户的原始需求。
2. 将需求分解成具体的开发任务。
3. 依次调度代码实现员、审查员、测试员等工作。
4. 整合最终成果。
请始终以项目管理的视角思考,关注任务依赖和整体进度。你的输出应该是清晰的下一步指令或决策。""",

    "analyst": """你是一个任务分析员。你的目标是将一个复杂的软件需求分解成原子化的、可执行的任务列表。
请按逻辑顺序列出任务,并考虑前后依赖关系。输出格式必须是严格的JSON:
{
  "tasks": [
    {"id": 1, "description": "任务1描述", "type": "backend|frontend|db|config|doc|test", "depends_on": []},
    {"id": 2, "description": "任务2描述", "type": "frontend", "depends_on": [1]}
  ]
}""",

    "architect": """你是一个软件架构师。基于当前任务和项目上下文,选择最合适的技术栈、设计核心模块和目录结构。
请给出简要的理由。输出格式:
{
  "tech_stack": {"frontend": "React", "backend": "Express.js", ...},
  "project_structure": ["/src", "/src/components", ...],
  "rationale": "选择React因为..."
}""",

    "coder": """你是一个资深{role}工程师。你的任务是根据详细描述和架构设计,编写高质量、可维护的代码。
请只输出代码本身,或严格按照以下JSON格式输出多个文件:
{
  "action": "write_code",
  "files": [
    {"path": "文件路径", "content": "文件内容"}
  ]
}
确保代码语法正确,遵循最佳实践。""",

    "reviewer": """你是一个严格的代码审查员。检查提供的代码,找出bug、坏味道、风格问题、安全漏洞和性能隐患。
对每个问题,指明文件、行号(如果可能)和具体建议。输出格式:
{
  "action": "code_review",
  "file_path": "被审查文件路径",
  "issues": [
    {"line": 10, "severity": "high|medium|low", "suggestion": "具体修改建议"}
  ],
  "approved": true/false
}
只有问题全部解决或仅为低风险建议时,才标记 approved=true。""",

    "tester": """你是一个测试工程师。为给定的代码功能编写全面的单元测试。
使用常见的测试框架(如Jest for JavaScript, pytest for Python)。输出格式:
{
  "action": "write_test",
  "file_path": "测试文件路径(应与原代码对应)",
  "content": "测试代码内容"
}""",

    "documenter": """你是一个技术文档工程师。为当前项目或模块编写清晰的使用说明、API文档或代码注释。
输出格式:
{
  "action": "write_doc",
  "files": [
    {"path": "README.md", "content": "..."},
    {"path": "src/module.js", "section": "header", "content": "// 注释..."}
  ]
}"""
}

注意, coder 的提示词中包含 {role} 占位符,我们可以在运行时填入 frontend , backend 等,实现更精细的分工。

3.2 构建项目状态管理器

我们需要一个中心化的对象来跟踪项目的一切。

import json
from typing import Dict, List, Any, Optional

class ProjectState:
    """管理项目状态的核心类"""
    def __init__(self, user_request: str):
        self.user_request = user_request
        self.tasks: List[Dict] = []  # 从 analyst 解析来的任务列表
        self.current_task_index: int = 0
        self.code_files: Dict[str, str] = {}  # 路径 -> 内容
        self.review_notes: List[Dict] = []
        self.test_files: Dict[str, str] = {}
        self.documentation: Dict[str, str] = {}
        self.architecture: Optional[Dict] = None
        self.conversation_history: List[Dict] = []  # 记录所有角色对话

    def add_code_file(self, path: str, content: str):
        """添加或更新代码文件"""
        self.code_files[path] = content

    def get_task(self) -> Optional[Dict]:
        """获取当前待处理的任务"""
        if self.current_task_index < len(self.tasks):
            return self.tasks[self.current_task_index]
        return None

    def complete_current_task(self):
        """标记当前任务完成,移向下一个"""
        if self.current_task_index < len(self.tasks):
            self.current_task_index += 1

    def to_context_string(self) -> str:
        """将项目状态序列化为字符串,供模型参考"""
        context = f"# 用户原始需求\n{self.user_request}\n\n"
        context += f"# 任务列表(已完成前{self.current_task_index}个)\n"
        for i, task in enumerate(self.tasks):
            status = "✅" if i < self.current_task_index else "⏳"
            context += f"{status} {task['description']} [类型:{task['type']}]\n"
        
        context += f"\n# 已生成代码文件\n"
        for path in self.code_files:
            context += f"- `{path}`\n"
        
        # 可以添加最近的审查意见等
        if self.review_notes:
            context += f"\n# 最新审查意见\n{json.dumps(self.review_notes[-1], indent=2, ensure_ascii=False)}\n"
        
        return context

3.3 实现团队协调员

这是最核心的类,它驱动整个流程。

class AgentTeamCoordinator:
    def __init__(self, llm_client, initial_request: str):
        self.llm = llm_client  # 假设这是我们已经封装好的模型调用客户端
        self.state = ProjectState(initial_request)
        self.role_prompts = ROLE_PROMPTS

    def run(self):
        """主运行循环"""
        print(f"开始处理需求:{self.state.user_request}")
        
        # 阶段1:需求分析与任务分解
        self._analyze_tasks()
        
        # 阶段2:技术架构设计(可选,对于简单任务可跳过)
        if len(self.state.tasks) > 1:  # 多任务项目才需要架构设计
            self._design_architecture()
        
        # 阶段3:循环执行每个任务
        while task := self.state.get_task():
            print(f"\n=== 正在处理任务 {self.state.current_task_index + 1}/{len(self.state.tasks)}: {task['description']} ===")
            
            # 根据任务类型分配具体的编码角色
            coder_role = self._map_task_type_to_coder(task['type'])
            
            # 子循环:编码 -> 审查 -> (可能)修改 -> 测试 -> 文档
            max_iterations = 3  # 防止无限循环
            for iteration in range(max_iterations):
                print(f"  迭代 {iteration + 1}: 编码...")
                # 3.3.1 编码
                success, coder_output = self._execute_role("coder", task, coder_role)
                if not success:
                    print("编码阶段失败,跳出任务。")
                    break
                    
                # 解析编码输出,更新代码文件
                self._parse_and_update_code(coder_output)
                
                print(f"  迭代 {iteration + 1}: 审查...")
                # 3.3.2 审查
                review_passed, review_output = self._execute_review(task)
                if review_passed:
                    print("  代码审查通过!")
                    # 审查通过,可进行测试和文档
                    self._execute_test(task)
                    self._execute_documentation(task)
                    self.state.complete_current_task()
                    break  # 跳出当前任务的迭代循环
                else:
                    print(f"  审查发现{len(review_output.get('issues', []))}个问题,进入下一轮修改。")
                    # 将审查意见加入上下文,下一轮编码时会看到
                    self.state.review_notes.append(review_output)
                    # 继续循环,进行下一轮编码(修改)
            else:
                print(f"警告:任务'{task['description']}'在{max_iterations}轮迭代后仍未通过审查。")
                self.state.complete_current_task()  # 强制标记完成,继续下一个任务
        
        # 阶段4:项目整合与最终交付
        self._deliver_project()
    
    def _analyze_tasks(self):
        """调用分析师角色分解任务"""
        analyst_prompt = self.role_prompts["analyst"]
        # 将用户需求作为用户输入
        user_message = f"请将以下需求分解为开发任务:\n{self.state.user_request}"
        
        full_prompt = f"{analyst_prompt}\n\n{user_message}"
        response = self.llm.chat(full_prompt)
        
        try:
            # 尝试从响应中解析JSON
            parsed = json.loads(response)
            self.state.tasks = parsed.get("tasks", [])
            print(f"任务分解完成,共{len(self.state.tasks)}个子任务。")
        except json.JSONDecodeError:
            print("警告:分析师返回了非JSON格式,尝试手动提取...")
            # 这里可以添加一些启发式规则来提取任务,作为降级方案
            self.state.tasks = [{"id": 1, "description": "实现核心功能", "type": "fullstack", "depends_on": []}]
    
    def _design_architecture(self):
        """调用架构师角色进行设计"""
        arch_prompt = self.role_prompts["architect"]
        context = f"项目需求:{self.state.user_request}\n初步任务列表:{self.state.tasks}"
        response = self.llm.chat(f"{arch_prompt}\n\n{context}")
        try:
            self.state.architecture = json.loads(response)
            print(f"架构设计完成:{self.state.architecture.get('tech_stack', {})}")
        except:
            print("架构设计解析失败,使用默认设置。")
    
    def _map_task_type_to_coder(self, task_type: str) -> str:
        """将任务类型映射到具体的编码员角色"""
        mapping = {
            "frontend": "前端",
            "backend": "后端",
            "db": "数据库",
            "fullstack": "全栈"
        }
        return mapping.get(task_type, "全栈")
    
    def _execute_role(self, role_name: str, task: Dict, specialisation: str = None) -> (bool, Any):
        """执行一个特定角色的工作"""
        base_prompt = self.role_prompts[role_name]
        if role_name == "coder" and specialisation:
            base_prompt = base_prompt.format(role=specialisation)
        
        # 构建当前上下文
        context = self.state.to_context_string()
        current_task_desc = f"当前具体任务:{task['description']}\n任务类型:{task['type']}"
        
        # 如果是编码员,并且有架构设计,也提供
        if role_name == "coder" and self.state.architecture:
            arch_info = f"\n架构设计:{json.dumps(self.state.architecture, ensure_ascii=False)}"
            current_task_desc += arch_info
        
        # 如果有未解决的审查意见,也提供给编码员(用于修改)
        if role_name == "coder" and self.state.review_notes:
            last_review = self.state.review_notes[-1]
            review_info = f"\n上一轮审查意见(请据此修改):\n{json.dumps(last_review, ensure_ascii=False, indent=2)}"
            current_task_desc += review_info
        
        user_message = f"{context}\n\n{current_task_desc}\n\n请开始你的工作:"
        
        full_prompt = f"{base_prompt}\n\n{user_message}"
        response = self.llm.chat(full_prompt)
        
        # 记录到对话历史
        self.state.conversation_history.append({
            "role": role_name,
            "prompt": full_prompt,
            "response": response
        })
        
        return True, response
    
    def _parse_and_update_code(self, coder_output: str):
        """解析编码员的输出,更新项目状态中的代码文件"""
        # 首先尝试解析为JSON
        try:
            result = json.loads(coder_output)
            if result.get("action") == "write_code" and "files" in result:
                for file_info in result["files"]:
                    path = file_info.get("path")
                    content = file_info.get("content")
                    if path and content is not None:
                        self.state.add_code_file(path, content)
                        print(f"    生成/更新文件:{path}")
                return
        except json.JSONDecodeError:
            pass
        
        # 如果不是标准JSON,可能编码员直接输出了代码文本
        # 我们可以尝试一些启发式规则,比如根据任务描述猜测文件名
        # 这里简化处理:存入一个默认文件
        default_path = f"task_{self.state.current_task_index + 1}_code.txt"
        self.state.add_code_file(default_path, coder_output)
        print(f"    非标准输出,已保存至:{default_path}")
    
    def _execute_review(self, task: Dict) -> (bool, Dict):
        """执行代码审查"""
        # 获取当前任务可能相关的代码文件(这里简化:审查所有代码)
        if not self.state.code_files:
            return True, {"action": "code_review", "approved": True, "note": "暂无代码可审查"}  # 没有代码,直接通过
        
        # 构建审查上下文:展示所有相关代码
        code_context = "待审查的代码文件:\n"
        for path, content in self.state.code_files.items():
            code_context += f"\n--- 文件:{path} ---\n{content}\n"
        
        review_prompt = self.role_prompts["reviewer"]
        user_message = f"{code_context}\n\n请审查以上代码。"
        
        response = self.llm.chat(f"{review_prompt}\n\n{user_message}")
        
        try:
            review_result = json.loads(response)
            approved = review_result.get("approved", False)
            return approved, review_result
        except:
            # 如果解析失败,保守起见,认为审查未通过
            return False, {"action": "code_review", "approved": False, "issues": [{"suggestion": "审查员返回了无法解析的格式。"}]}
    
    def _execute_test(self, task: Dict):
        """为当前任务生成测试"""
        if "test" not in task.get("type", ""):  # 如果任务类型不是测试,且我们想为它生成测试
            test_prompt = self.role_prompts["tester"]
            code_context = "需要编写测试的代码:\n"
            for path, content in self.state.code_files.items():
                code_context += f"\n--- 文件:{path} ---\n{content}\n"
            
            response = self.llm.chat(f"{test_prompt}\n\n{code_context}")
            try:
                test_result = json.loads(response)
                if test_result.get("action") == "write_test":
                    path = test_result.get("file_path", f"test_task_{self.state.current_task_index}.js")
                    content = test_result.get("content")
                    self.state.test_files[path] = content
                    print(f"    生成测试文件:{path}")
            except:
                print("    测试生成失败或格式错误。")
    
    def _execute_documentation(self, task: Dict):
        """为当前任务或整体项目生成文档"""
        # 简化:在最后一个任务完成后,生成整体项目文档
        if self.state.current_task_index == len(self.state.tasks) - 1:  # 如果是最后一个任务
            doc_prompt = self.role_prompts["documenter"]
            project_context = f"项目需求:{self.state.user_request}\n\n已生成代码:{list(self.state.code_files.keys())}"
            response = self.llm.chat(f"{doc_prompt}\n\n{project_context}")
            try:
                doc_result = json.loads(response)
                if doc_result.get("action") == "write_doc":
                    for file_info in doc_result.get("files", []):
                        path = file_info.get("path")
                        content = file_info.get("content")
                        if path and content:
                            self.state.documentation[path] = content
                            print(f"    生成文档:{path}")
            except:
                print("    文档生成失败或格式错误。")
    
    def _deliver_project(self):
        """交付最终项目"""
        print("\n" + "="*50)
        print("项目开发完成!交付物如下:")
        print("="*50)
        print("\n【生成的代码文件】")
        for path, content in self.state.code_files.items():
            print(f"- {path} ({len(content)} 字符)")
            # 这里可以实际写入文件系统
            # with open(path, 'w', encoding='utf-8') as f:
            #     f.write(content)
        
        print("\n【生成的测试文件】")
        for path, content in self.state.test_files.items():
            print(f"- {path}")
        
        print("\n【生成的文档】")
        for path, content in self.state.documentation.items():
            print(f"- {path}")
        
        print("\n【项目总结】")
        print(f"原始需求:{self.state.user_request}")
        print(f"共处理 {len(self.state.tasks)} 个子任务。")
        if self.state.architecture:
            print(f"采用技术栈:{self.state.architecture.get('tech_stack', {})}")

3.4 集成与使用示例

最后,我们需要将 AgentTeamCoordinator 集成到我们现有的 learn-claude-code 主程序中。假设我们有一个命令行接口。

# 在主程序中的调用示例
def main():
    llm_client = ClaudeCodeAgent()  # 假设这是我们已经实现的智能体
    
    print("欢迎使用 ClaudeCode 智能体团队模式!")
    user_request = input("请输入您的开发需求:")
    
    if not user_request.strip():
        print("需求不能为空。")
        return
    
    # 简单判断:如果需求复杂(例如包含多个功能点或指定了“项目”),则使用团队模式
    use_team_mode = len(user_request.split()) > 10 or "项目" in user_request or "应用" in user_request
    
    if use_team_mode:
        print("检测到复杂需求,启动智能体团队协作模式...")
        team = AgentTeamCoordinator(llm_client, user_request)
        team.run()
    else:
        print("使用单智能体模式处理...")
        # 原有的单智能体处理逻辑
        response = llm_client.chat(user_request)
        print(response)

if __name__ == "__main__":
    main()

4. 实战中的挑战、优化与避坑指南

上面的代码框架展示了一个基础的智能体团队实现,但在实际运行中,你会遇到不少挑战。下面是我在实验过程中总结的一些关键点和优化方向。

4.1 挑战一:上下文长度与成本控制

这是最现实的问题。团队协作意味着多次的模型调用和不断增长的上下文。一个中等复杂度的项目,来回几次迭代,上下文很容易超过 Claude/GPT 的窗口限制(如 128K/200K)。

优化策略:

  1. 选择性上下文 :不要每次都把整个项目历史喂给模型。对于“编码员”,主要提供当前任务描述、架构设计、以及 最近一次相关的审查意见 。对于“审查员”,只提供需要审查的 特定文件内容 ,而不是全部代码。
  2. 总结与摘要 :定期对已完成的对话或代码变更进行总结。例如,在任务完成后,用一小段话总结这个任务实现了什么,有什么关键设计,替代原始的长篇对话历史。
  3. 分层任务分解 :如果项目非常大,不要试图一次性分解出所有任务。可以先进行“高层规划”,生成几个大的模块(如“用户认证模块”、“数据看板模块”),然后针对每个模块再启动一个新的、独立的团队协作流程,每个流程有自己的、较短的生命周期上下文。
  4. 使用更便宜的模型 :并非所有角色都需要最强大的模型。例如,“文档工程师”和部分“代码审查”(检查简单风格)的工作,可以使用更便宜、速度更快的模型(如 Claude Haiku, GPT-3.5-Turbo)来完成,以降低成本。

4.2 挑战二:角色扮演的“幻觉”与一致性

模型可能会“忘记”自己当前扮演的角色,或者在不同角色的思维模式间切换不彻底。例如,“编码员”可能突然开始写审查意见。

优化策略:

  1. 强化系统提示词 :在每个角色的提示词开头,用非常强烈、清晰的语言强调角色。例如:“你 现在是且仅仅是 一个专注于编写 React 前端组件的工程师。你 绝对不能 进行代码审查或设计架构。你的 唯一目标 是根据要求产出代码。”
  2. 输出格式强制约束 :如前所述,强制要求 JSON 等结构化输出,并在提示词中明确说明“你必须且只能以以下 JSON 格式回应”。如果模型返回了错误格式,我们的程序可以解析失败,并发送一个修正指令(如“请严格遵循指定的 JSON 格式重新回答”)。
  3. 会话隔离 :为每个角色使用独立的“会话”(即不共享对话历史)。每次调用都是一个全新的、只包含该系统提示词和当前任务上下文的对话。这能最大程度避免角色混淆,但代价是模型无法从之前的交互中学习(不过在我们的流程中,状态是由 ProjectState 管理的,所以影响不大)。

4.3 挑战三:任务分解与依赖管理的智能化

我们的简单实现中,任务分解依赖一次 LLM 调用,且依赖关系是静态的。现实中,任务依赖可能动态变化,或者分解得不够好。

优化策略:

  1. 迭代式任务规划 :不要一次性分解所有任务。可以先让“协调员”生成一个 高层计划 (High-level Plan),然后每完成一个任务,都重新评估剩余任务和依赖关系,动态调整计划。这更接近人类的敏捷开发。
  2. 依赖检测与死锁预防 :在程序逻辑中加入简单的依赖检查。如果任务 A 依赖任务 B,但任务 B 又(直接或间接)依赖任务 A,就形成了死锁。协调员需要能检测这种循环依赖,并重新规划或请求人工干预。
  3. 任务粒度控制 :通过提示词引导分析师,将任务分解到“一个熟练开发者 2-4 小时能完成”的粒度。太粗了无法并行且容易出错,太细了则管理开销巨大。

4.4 一个高级技巧:让协调员“自我反思”

我们可以让“项目协调员”角色在关键节点(如一个任务完成后、或遇到审查多次不通过时)进行“自我反思”。它的提示词可以增加:

“在做出下一步决策前,请先简要反思:当前的项目进度是否健康?是否有任务卡住?代码质量趋势如何?基于此反思,调整你接下来的调度策略。”

然后,让协调员输出一个包含 reflection next_action 的 JSON。这能引入一定的元认知能力,让流程更健壮。

4.5 避坑实操心得

  1. 从简单任务开始 :不要一开始就让它构建一个“完整的电商平台”。从“创建一个有增删改查的待办事项 API”开始。验证流程跑通,再增加复杂度。
  2. 设置迭代上限和超时 :代码审查-修改循环必须设置最大迭代次数(如上述代码中的 max_iterations ),否则一个无法修复的 bug 可能导致无限循环,消耗大量 token 和费用。
  3. 人工检查点 :在关键节点(如架构设计确认、任务分解清单)设置人工确认环节。让用户看一眼,说“可以”,再继续。这能防止项目跑偏。
  4. 持久化项目状态 :一定要把 ProjectState 对象定期保存到文件(如 JSON)。这样如果程序中断或出错,你可以从中断点恢复,而不是重头开始。
  5. 日志至关重要 :详细记录每个角色的输入和输出。这不仅是调试的需要,更是你优化提示词、理解模型行为的宝贵数据。 conversation_history 就是这个目的。

5. 超越编码:智能体团队模式的泛化思考

我们虽然以 learn-claude-code 项目为例,实现了针对软件开发的智能体团队,但这一模式的潜力远不止于此。其核心范式—— “一个协调者 + 多个专业化执行者” ——可以迁移到无数领域。

  • 内容创作团队 :协调者接收主题大纲,调度“调研员”收集资料、“撰稿人”撰写初稿、“编辑”进行润色和校对、“排版员”进行格式优化。
  • 数据分析团队 :协调者接收分析需求,调度“数据清洗员”处理原始数据、“分析师 A”进行描述性统计、“分析师 B”建立预测模型、“可视化专家”制作图表。
  • 客服与支持团队 :协调者解读用户问题,如果是技术问题转给“技术客服”,如果是账单问题转给“财务客服”,如果是使用咨询转给“产品专家”,最后整合答案回复用户。
  • 游戏开发 :协调者根据设计文档,调度“剧情编剧”、“关卡设计师”、“角色美术”、“音效师”等。

实现这些泛化团队的关键,在于为你领域内的每个“角色”精心设计其 系统提示词 输入输出规范 以及它们之间的 协作协议 AgentTeamCoordinator 的框架可以复用,你只需要更换 ROLE_PROMPTS 字典和 ProjectState 中跟踪的数据类型即可。

learn-claude-code 中实现 Agent Teams,与其说是一个功能,不如说是一次对 AI 智能体如何模拟复杂、结构化工作流的深度探索。它迫使我们去思考如何将模糊的指令转化为清晰的步骤,如何让多个“思维链”有序协作,以及如何将人的项目管理智慧编码进提示词和流程里。这个过程里踩的每一个坑,都是对 AI 协作本质更深入的理解。

更多推荐