Claude Code 多智能体协作实战：用子任务编排构建复杂工程项目的完整指南

Claude Code 多智能体协作实战：用子任务编排构建复杂工程项目的完整指南

> 从单体 Prompt 到多 Agent 协作，手把手教你用 Claude Code 的 Agent 工具链完成真实项目

一、为什么需要多智能体协作？

随着项目复杂度的提升，单一 AI 对话窗口的局限性越来越明显。当你需要同时处理代码编写、测试生成、文档撰写、代码审查等多个环节时，一个 Agent 往往会出现上下文丢失、任务遗忘等问题。

Claude Code 提供了一套完整的 Agent 工具链（Task 工具），支持在主对话中派生子 Agent 独立执行子任务，主 Agent 负责编排和汇总。这种多智能体协作模式，本质上是一种主从式任务分发架构。

核心优势

二、Claude Code 子 Agent 架构详解

Claude Code 中的 Task 工具支持三种子 Agent 类型：

2.1 基本调用原理

当你在 Claude Code 中描述一个复杂任务时，Claude 会自动（或被指示）将任务拆解为多个子任务，每个子任务通过 Task 工具派发给子 Agent：

主 Agent (编排层)
├── 子 Agent A (general_purpose_task) → 编写核心业务逻辑
├── 子 Agent B (general_purpose_task) → 编写测试用例
├── 子 Agent C (Explore) → 分析现有代码结构
└── 主 Agent → 汇总结果，处理集成

2.2 Claude Code Hooks 实现自动化工作流

Claude Code 支持 Hooks 机制，可以在特定事件触发时自动执行脚本。这是构建自动化流水线的关键。

项目配置文件 .claude/settings.json：

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Write|Edit",
        "hooks": [
          {
            "type": "command",
            "command": "python3 scripts/lint_check.py \"$FILE\""
          }
        ]
      }
    ],
    "PostToolUse": [
      {
        "matcher": "Write|Edit",
        "hooks": [
          {
            "type": "command",
            "command": "cd /path/to/project && pytest tests/ -x -q 2>/dev/null || echo 'Tests not found'"
          }
        ]
      }
    ],
    "Notification": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "osascript -e 'display notification \"Task completed\" with title \"Claude Code\"'"
          }
        ]
      }
    ]
  }
}

三、实战案例：构建一个 REST API 项目

我们将用多 Agent 协作的方式，从零构建一个带认证的 REST API 服务。

3.1 项目结构

fastapi-auth-service/
├── app/
│   ├── __init__.py
│   ├── main.py          # 主入口
│   ├── auth/
│   │   ├── __init__.py
│   │   ├── router.py    # 认证路由
│   │   ├── service.py   # 认证服务逻辑
│   │   └── models.py    # 数据模型
│   ├── users/
│   │   ├── __init__.py
│   │   ├── router.py
│   │   ├── service.py
│   │   └── models.py
│   └── database.py
├── tests/
│   ├── test_auth.py
│   ├── test_users.py
│   └── conftest.py
├── .claude/
│   └── settings.json
├── requirements.txt
├── Dockerfile
└── README.md

3.2 步骤一：用 Explore Agent 分析需求

首先让 Explore Agent 理清项目依赖和最佳实践：

【子任务指令示例 - 发给 Explore Agent】
调研 FastAPI + JWT 认证的最佳实践方案，包括：
1. 推荐的依赖库版本（fastapi, python-jose, passlib, sqlalchemy）
2. 项目目录结构最佳实践
3. JWT token 签发与验证的安全配置
4. 密码哈希方案选择（bcrypt vs argon2）
返回完整的调研报告。

3.3 步骤二：用 general_purpose_task Agent 编写核心代码

数据模型（app/auth/models.py）：

from datetime import datetime
from sqlalchemy import Column, String, DateTime, Boolean
from sqlalchemy.orm import declarative_base
import uuid

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(255), unique=True, nullable=False, index=True)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)
    is_superuser = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

class RefreshToken(Base):
    __tablename__ = "refresh_tokens"

    id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    user_id = Column(String(36), nullable=False, index=True)
    token = Column(String(512), nullable=False, unique=True)
    expires_at = Column(DateTime, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    is_revoked = Column(Boolean, default=False)

认证服务（app/auth/service.py）：

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
import secrets
import uuid

SECRET_KEY = "your-secret-key-change-in-production"  # 生产环境请从环境变量读取
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class AuthService:
    def __init__(self, secret_key: str = SECRET_KEY):
        self.secret_key = secret_key

    def hash_password(self, plain_password: str) -> str:
        """密码哈希"""
        return pwd_context.hash(plain_password)

    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
        """密码验证"""
        return pwd_context.verify(plain_password, hashed_password)

    def create_access_token(self, user_id: str, is_superuser: bool = False) -> str:
        """创建 JWT Access Token"""
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        payload = {
            "sub": user_id,
            "is_superuser": is_superuser,
            "type": "access",
            "exp": expire,
            "jti": str(uuid.uuid4())
        }
        return jwt.encode(payload, self.secret_key, algorithm=ALGORITHM)

    def create_refresh_token(self) -> dict:
        """创建 Refresh Token（存数据库的 opaque token + JWT 编码）"""
        token_value = secrets.token_urlsafe(48)
        return {
            "token": token_value,
            "expires_at": datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
            "jti": str(uuid.uuid4())
        }

    def decode_token(self, token: str) -> Optional[dict]:
        """解码并验证 JWT Token"""
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=[ALGORITHM])
            return payload
        except JWTError:
            return None

# 全局实例
auth_service = AuthService()

认证路由（app/auth/router.py）：

from fastapi import APIRouter, HTTPException, Depends, status
from pydantic import BaseModel, EmailStr
from typing import Optional

router = APIRouter(prefix="/api/auth", tags=["认证"])

class RegisterRequest(BaseModel):
    username: str
    email: EmailStr
    password: str

class LoginRequest(BaseModel):
    username: str
    password: str

class TokenResponse(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"

class RefreshRequest(BaseModel):
    refresh_token: str

@router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED)
async def register(request: RegisterRequest):
    """用户注册"""
    from .service import auth_service
    from ..database import get_db

    db = next(get_db())

    # 检查用户名是否已存在
    existing_user = db.query(User).filter(
        (User.username == request.username) | (User.email == request.email)
    ).first()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="用户名或邮箱已存在"
        )

    # 创建用户
    user = User(
        username=request.username,
        email=request.email,
        hashed_password=auth_service.hash_password(request.password)
    )
    db.add(user)
    db.commit()
    db.refresh(user)

    # 生成 token
    access_token = auth_service.create_access_token(user.id)
    refresh_data = auth_service.create_refresh_token()

    return TokenResponse(
        access_token=access_token,
        refresh_token=refresh_data["token"]
    )

@router.post("/login", response_model=TokenResponse)
async def login(request: LoginRequest):
    """用户登录"""
    from .service import auth_service
    from ..database import get_db

    db = next(get_db())
    user = db.query(User).filter(User.username == request.username).first()

    if not user or not auth_service.verify_password(request.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"}
        )

    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="账户已被禁用"
        )

    access_token = auth_service.create_access_token(user.id, user.is_superuser)
    refresh_data = auth_service.create_refresh_token()

    return TokenResponse(
        access_token=access_token,
        refresh_token=refresh_data["token"]
    )

@router.post("/refresh", response_model=TokenResponse)
async def refresh_token(request: RefreshRequest):
    """刷新 Token"""
    from .service import auth_service
    from ..database import get_db

    db = next(get_db())
    stored_token = db.query(RefreshToken).filter(
        RefreshToken.token == request.refresh_token,
        RefreshToken.is_revoked == False,
        RefreshToken.expires_at > datetime.utcnow()
    ).first()

    if not stored_token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Refresh Token 无效或已过期"
        )

    # 撤销旧 token
    stored_token.is_revoked = True
    db.commit()

    # 生成新 token
    access_token = auth_service.create_access_token(stored_token.user_id)
    refresh_data = auth_service.create_refresh_token()

    return TokenResponse(
        access_token=access_token,
        refresh_token=refresh_data["token"]
    )

3.4 步骤三：用另一个子 Agent 编写测试

【子任务指令示例 - 发给 general_purpose_task Agent】
基于 app/auth/router.py 和 app/auth/service.py 编写完整的 pytest 测试：
1. test_auth.py - 覆盖注册、登录、Token刷新、错误处理
2. conftest.py - 提供测试数据库 fixture 和测试客户端
3. 使用 SQLite 内存数据库进行测试
4. 确保测试覆盖率 > 80%

测试文件（tests/test_auth.py）：

import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.database import engine, Base, get_db
from sqlalchemy.orm import sessionmaker

TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)

@pytest.fixture(autouse=True)
def setup_db():
    """每个测试前重建数据库"""
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)

def test_register_success():
    """测试正常注册"""
    response = client.post("/api/auth/register", json={
        "username": "testuser",
        "email": "test@example.com",
        "password": "SecurePass123!"
    })
    assert response.status_code == 201
    data = response.json()
    assert "access_token" in data
    assert "refresh_token" in data
    assert data["token_type"] == "bearer"

def test_register_duplicate_username():
    """测试重复用户名注册"""
    client.post("/api/auth/register", json={
        "username": "dupuser",
        "email": "dup1@example.com",
        "password": "SecurePass123!"
    })
    response = client.post("/api/auth/register", json={
        "username": "dupuser",
        "email": "dup2@example.com",
        "password": "SecurePass123!"
    })
    assert response.status_code == 400
    assert "已存在" in response.json()["detail"]

def test_login_success():
    """测试正常登录"""
    # 先注册
    client.post("/api/auth/register", json={
        "username": "loginuser",
        "email": "login@example.com",
        "password": "LoginPass123!"
    })
    # 再登录
    response = client.post("/api/auth/login", json={
        "username": "loginuser",
        "password": "LoginPass123!"
    })
    assert response.status_code == 200
    assert "access_token" in response.json()

def test_login_wrong_password():
    """测试错误密码"""
    client.post("/api/auth/register", json={
        "username": "wronguser",
        "email": "wrong@example.com",
        "password": "CorrectPass123!"
    })
    response = client.post("/api/auth/login", json={
        "username": "wronguser",
        "password": "WrongPass123!"
    })
    assert response.status_code == 401

def test_login_nonexistent_user():
    """测试不存在的用户"""
    response = client.post("/api/auth/login", json={
        "username": "ghost",
        "password": "NoPass123!"
    })
    assert response.status_code == 401

def test_refresh_token():
    """测试 Token 刷新"""
    reg_resp = client.post("/api/auth/register", json={
        "username": "refreshuser",
        "email": "refresh@example.com",
        "password": "RefreshPass123!"
    })
    refresh_token = reg_resp.json()["refresh_token"]

    response = client.post("/api/auth/refresh", json={
        "refresh_token": refresh_token
    })
    assert response.status_code == 200
    new_tokens = response.json()
    assert "access_token" in new_tokens
    # 旧 token 应该失效
    old_response = client.post("/api/auth/refresh", json={
        "refresh_token": refresh_token
    })
    assert old_response.status_code == 401

def test_register_short_password():
    """测试弱密码拒绝"""
    response = client.post("/api/auth/register", json={
        "username": "weakuser",
        "email": "weak@example.com",
        "password": "123"
    })
    # 密码过短应该被 Pydantic 验证拦截
    assert response.status_code == 422

四、Claude Code 高级技巧

4.1 Claude Code 命令行快捷操作

# 在项目根目录初始化 Claude Code
claude

# 直接执行单条指令（非交互模式）
claude -p "分析 src/ 目录下的代码质量，输出改进建议"

# 从文件读取 prompt 执行
claude -p "$(cat prompts/refactor-task.md)"

# 使用特定模型
claude --model claude-sonnet-4-20250514

# 查看 Claude Code 设置
claude config list

4.2 MCP 工具集成实战

Claude Code 原生支持 MCP（Model Context Protocol），可以连接外部数据源。以下是一个连接 GitHub MCP Server 的配置：

.claude/settings.json：

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_TOKEN": "${GITHUB_TOKEN}"
      }
    },
    "database": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres", "postgresql://localhost/mydb"]
    }
  }
}

配置后，Claude Code 的 Agent 可以直接查询 GitHub Issues、PR、数据库数据等。

4.3 上下文管理技巧

Claude Code 支持通过 / 命令快速管理上下文：

# 在 Claude Code 交互中：
/add file.py           # 将文件添加到上下文
/add src/**/*.py      # 批量添加目录
/remove file.py       # 移除文件
/clear                 # 清除对话历史
/compact               # 压缩对话历史（保留关键信息）
/model                 # 切换模型

4.4 多文件编辑的自动化工作流

当你需要批量修改多个文件时，Claude Code 的 Agent 可以自动编排：

【给 Claude Code 的指令】
我需要将项目中所有的 print() 调试语句替换为 logging 调用。
请执行以下步骤：
1. 先用 Explore Agent 搜索所有包含 print() 的 Python 文件
2. 用 general_purpose_task Agent 逐文件替换，在每个文件头部添加 logging 配置
3. 运行测试确认没有破坏功能
4. 输出修改清单

五、生产部署方案

5.1 Docker 部署配置

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# docker-compose.yml
version: '3.8'
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - SECRET_KEY=${SECRET_KEY}
      - DATABASE_URL=postgresql://user:pass@db:5432/authdb
    depends_on:
      - db
  db:
    image: postgres:15-alpine
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: authdb
    volumes:
      - pgdata:/var/lib/postgresql/data
volumes:
  pgdata:

5.2 Nginx 反向代理配置

server {
    listen 80;
    server_name api.example.com;
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl http2;
    server_name api.example.com;

    ssl_certificate /etc/nginx/ssl/api.example.com.pem;
    ssl_certificate_key /etc/nginx/ssl/api.example.com.key;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # 速率限制
        limit_req zone=api burst=20 nodelay;
    }

    location /docs {
        proxy_pass http://127.0.0.1:8000/docs;
        # 仅内网可访问文档
        allow 10.0.0.0/8;
        deny all;
    }
}

六、常见问题 FAQ

Q1: Claude Code 子 Agent 数量有限制吗？

Claude Code 建议同时运行不超过 3 个子 Agent。每个子 Agent 独立消耗 token，过多并行会导致 API 配额快速消耗。建议将任务按优先级排序，关键路径上的任务优先执行。

Q2: 如何避免子 Agent 之间的文件冲突？

在 Claude Code 的 Task 工具中，可以通过参数指定子 Agent 的工作目录和允许操作的范围。建议不同子 Agent 操作不同目录或文件，最后由主 Agent 进行合并。

Q3: Hooks 执行失败会影响主任务吗？

默认情况下，Hooks 执行失败不会阻断主任务。Claude Code 会记录 Hook 的输出和错误，但会继续执行后续操作。如果你需要严格的行为（如 lint 检查不通过则阻止写入），可以在 Hook 脚本中返回非零退出码。

Q4: 如何在 Claude Code 中使用自定义的 MCP Server？

在 .claude/settings.json 中配置 mcpServers 字段，指定命令、参数和环境变量。Claude Code 会在启动时自动连接这些 Server，Agent 执行时即可调用 Server 提供的工具。

Q5: Claude Code 支持哪些编程语言？

Claude Code 支持所有主流编程语言，包括 Python、JavaScript/TypeScript、Go、Rust、Java、C/C++ 等。其子 Agent 具备完整的代码生成、编辑和执行能力。

七、总结

Claude Code 的多智能体协作机制为复杂项目开发提供了全新的工作范式。通过合理的任务拆解和 Agent 编排，你可以：

核心要点：