子Agent概述#

子Agent是Claude Code中的一个重要概念，它允许主Agent将复杂任务分解为多个子任务，并委托给专门的子Agent来处理。通过子Agent机制，Claude Code可以更高效地处理复杂的编程任务，提高任务完成的质量和速度。

子Agent的概念#

1. 什么是子Agent#

子Agent是主Agent的辅助实体，具有以下特点：

• 专业化: 每个子Agent专注于特定的任务领域
• 独立性: 子Agent可以独立完成任务，不依赖主Agent的持续干预
• 协作性: 多个子Agent可以协同工作，完成复杂的任务
• 可复用性: 子Agent可以在不同的任务中被重复使用

2. 子Agent vs 主Agent#

子Agent的类型#

1. 代码生成子Agent#

示例：代码生成子Agent

用户请求： "创建一个用户认证系统的代码生成子Agent"

Claude Code 生成的子Agent：

python
复制
````python

classCodeGenerationAgent:"""代码生成子Agent"""
def__init__(self, context):self.context = context
self.templates ={}self.best_practices ={}
defgenerate_auth_system(self, requirements):"""生成认证系统代码"""# 分析需求analysis = self._analyze_requirements(requirements)
# 选择模板template = self._select_template(analysis)
# 生成代码code = self._generate_code(template, analysis)
# 应用最佳实践code = self._apply_best_practices(code)
# 验证代码validation = self._validate_code(code)
return{'code': code,'validation': validation,'documentation': self._generate_documentation(code)}
def_analyze_requirements(self, requirements):"""分析需求"""analysis ={'auth_methods':[],'security_level':'medium','features':[]}
if'jwt'in requirements:analysis['auth_methods'].append('jwt')
if'oauth'in requirements:analysis['auth_methods'].append('oauth')
if'2fa'in requirements:analysis['features'].append('2fa')
if'session'in requirements:analysis['auth_methods'].append('session')
return analysis

def_select_template(self, analysis):"""选择模板"""if'jwt'in analysis['auth_methods']:return'jwt_auth_template'elif'oauth'in analysis['auth_methods']:return'oauth_auth_template'else:return'basic_auth_template'
def_generate_code(self, template, analysis):"""生成代码"""if template =='jwt_auth_template':return self._generate_jwt_auth(analysis)elif template =='oauth_auth_template':return self._generate_oauth_auth(analysis)else:return self._generate_basic_auth(analysis)
def_generate_jwt_auth(self, analysis):"""生成JWT认证代码"""code ='''
from flask import Flask, request, jsonify
from flask_jwt_extended import(JWTManager, create_access_token,jwt_required, get_jwt_identity
)from datetime import timedelta
import bcrypt

app = Flask(__name__)app.config['JWT_SECRET_KEY']='your-secret-key'app.config['JWT_ACCESS_TOKEN_EXPIRES']= timedelta(hours=24)
jwt = JWTManager(app)
classAuthService:"""认证服务"""
def__init__(self, db):self.db = db

defregister(self, username, password):"""用户注册"""# 检查用户名是否存在if self._user_exists(username):raise ValueError("Username already exists")
# 哈希密码password_hash = bcrypt.hashpw(password.encode('utf-8'),bcrypt.gensalt()).decode('utf-8')
# 创建用户user = User(username=username, password_hash=password_hash)self.db.session.add(user)self.db.session.commit()
return user

deflogin(self, username, password):"""用户登录"""# 查找用户user = User.query.filter_by(username=username).first()
ifnot user:raise ValueError("Invalid username or password")
# 验证密码ifnot bcrypt.checkpw(password.encode('utf-8'),user.password_hash.encode('utf-8')):raise ValueError("Invalid username or password")
# 生成访问令牌access_token = create_access_token(identity=user.id)
return{'access_token': access_token,'user':{'id': user.id,'username': user.username
}}
def_user_exists(self, username):

"""检查用户名是否存在""" return User.query.filter_by(username=username).first() is not None ''' return code

def _apply_best_practices(self, code): """应用最佳实践"""

添加错误处理

code = self._add_error_handling(code)

添加日志记录

code = self._add_logging(code)

添加输入验证

code = self._add_input_validation(code)

return code

def _validate_code(self, code): """验证代码""" validation = { 'syntax': True, 'security': True, 'best_practices': True, 'issues': [] }

检查语法

try: compile(code, '', 'exec') except SyntaxError as e: validation['syntax'] = False validation['issues'].append(f"Syntax error: {e}")

检查安全性

if 'SECRET_KEY' in code and 'your-secret-key' in code: validation['security'] = False validation['issues'].append("Default secret key detected")

return validation

def _generate_documentation(self, code): """生成文档""" return { 'description': 'JWT-based authentication system', 'endpoints': [ { 'path': '/register', 'method': 'POST', 'description': 'Register a new user', 'parameters': { 'username': 'string (required)', 'password': 'string (required)' } }, { 'path': '/login', 'method': 'POST', 'description': 'Login with username and password', 'parameters': { 'username': 'string (required)', 'password': 'string (required)' } } ], 'security': { 'password_hashing': 'bcrypt', 'token': 'JWT', 'token_expiry': '24 hours' } }

bash
复制
> - 专注于代码生成
> - 包含模板库和最佳实践
> - 自动验证生成的代码
> - 生成配套文档

2. 代码审查子Agent#

示例：代码审查子Agent

用户请求： "创建一个代码审查子Agent" Claude Code 生成的子Agent：

python
复制
`python

classCodeReviewAgent:"""代码审查子Agent"""
def__init__(self, context):self.context = context
self.rules = self._load_review_rules()self.patterns = self._load_antipatterns()
defreview_code(self, code, language='python'):"""审查代码"""review ={'summary':{},'issues':[],'suggestions':[],'metrics':{}}
# 执行静态分析static_analysis = self._static_analysis(code, language)review['issues'].extend(static_analysis['issues'])
# 检查代码模式pattern_analysis = self._check_patterns(code, language)review['issues'].extend(pattern_analysis['issues'])review['suggestions'].extend(pattern_analysis['suggestions'])
# 计算代码指标metrics = self._calculate_metrics(code, language)review['metrics']= metrics

# 生成总结review['summary']= self._generate_summary(review)
return review

def_static_analysis(self, code, language):"""静态分析"""issues =[]
# 检查未使用的导入unused_imports = self._check_unused_imports(code, language)issues.extend(unused_imports)
# 检查未使用的变量unused_variables = self._check_unused_variables(code, language)issues.extend(unused_variables)
# 检查代码复杂度complexity_issues = self._check_complexity(code, language)issues.extend(complexity_issues)
# 检查安全问题security_issues = self._check_security(code, language)issues.extend(security_issues)
return{'issues': issues}
def_check_unused_imports(self, code, language):"""检查未使用的导入"""issues =[]
if language =='python':import ast
import re

tree = ast.parse(code)
# 获取所有导入imports =set()for node in ast.walk(tree):ifisinstance(node, ast.Import):for alias in node.names:imports.add(alias.name.split('.')[0])elifisinstance(node, ast.ImportFrom):if node.module:imports.add(node.module.split('.')[0])
# 检查是否使用for imp in imports:if imp notin code[code.find(imp):]:issues.append({'type':'unused_import','severity':'warning','message':f"Unused import: {imp}",'line': self._find_line(code, imp)})
return issues

def_check_complexity(self, code, language):"""检查代码复杂度"""issues =[]
if language =='python':import ast

tree = ast.parse(code)
for node in ast.walk(tree):ifisinstance(node,(ast.FunctionDef, ast.AsyncFunctionDef)):complexity = self._calculate_cyclomatic_complexity(node)
if complexity >10:issues.append({'type':'high_complexity','severity':'warning','message':f"Function '{node.name}' has high complexity ({complexity})",'line': node.lineno,'complexity': complexity
})
return issues

def_calculate_cyclomatic_complexity(self, node):"""计算圈复杂度"""complexity =1
for child in ast.walk(node):ifisinstance(child,(ast.If, ast.While, ast.For, ast.AsyncFor)):complexity +=1elifisinstance(child, ast.ExceptHandler):complexity +=1elifisinstance(child, ast.BoolOp):complexity +=len(child.values)-1
return complexity

def_check_security(self, code, language):"""检查安全问题"""issues =[]
# 检查硬编码密钥if'password'in code.lower()and'='in code:lines = code.split('\n')for i, line inenumerate(lines,1):if'password'in line.lower()and'"'in line:issues.append({'type':'security','severity':'error','message':'Possible hardcoded password detected','line': i
})
# 检查SQL注入风险if'execute'in code and'%'in code:lines = code.split('\n')for i, line inenumerate(lines,1):if'execute'in line and'%'in line andnot'param'in line.lower():issues.append({'type':'security','severity':'error','message':'Possible SQL injection vulnerability','line': i
})
return issues

def_check_patterns(self, code, language):"""检查代码模式"""issues =[]suggestions =[]
# 检查长函数if language =='python':import ast

tree = ast.parse(code)
for node in ast.walk(tree):ifisinstance(node,(ast.FunctionDef, ast.AsyncFunctionDef)):end_line = node.end_lineno ifhasattr(node,'end_lineno')else node.lineno
length = end_line - node.lineno

if length >50:issues.append({'type':'long_function','severity':'warning','message':f"Function '{node.name}' is too long ({length} lines)",'line': node.lineno,'length': length
})suggestions.append({'type':'refactor','message':f"Consider breaking down '{node.name}' into smaller functions",'line': node.lineno
})
return{'issues': issues,'suggestions': suggestions}
def_calculate_metrics(self, code, language):"""计算代码指标"""metrics ={}
lines = code.split('\n')
# 总行数metrics['total_lines']=len(lines)
# 代码行数（排除空行和注释）code_lines =[line for line in lines if line.strip()andnot line.strip().startswith('#')]metrics['code_lines']=len(code_lines)
# 注释行数comment_lines =[line for line in lines if line.strip().startswith('#')]metrics['comment_lines']=len(comment_lines)
# 空行数blank_lines =[line for line in lines ifnot line.strip()]metrics['blank_lines']=len(blank_lines)
# 注释率metrics['comment_ratio']= metrics['comment_lines']/ metrics['code_lines']if metrics['code_lines']>0else0
return metrics

def_generate_summary(self, review):"""生成审查总结"""summary ={'total_issues':len(review['issues']),'critical_issues':len([i for i in review['issues']if i['severity']=='error']),'warning_issues':len([i for i in review['issues']if i['severity']=='warning']),'total_suggestions':len(review['suggestions']),'code_quality':'good'}
# 评估代码质量if summary['critical_issues']>0:summary['code_quality']='poor'elif summary['warning_issues']>5:summary['code_quality']='fair'
return summary

```>**子Agent特点**:>- 专注于代码审查
>- 包含审查规则库
>- 检查代码模式和反模式
>- 计算代码指标

```
## 子Agent的协作
### 1. 串行协作
# 示例：子Agent串行协作用户请求：
"使用子Agent完成一个完整的开发任务"Claude Code 生成的协作流程：
````python
`python

classAgentOrchestrator:"""Agent协调器"""
def__init__(self, context):self.context = context
self.agents ={'code_generator': CodeGenerationAgent(context),'code_reviewer': CodeReviewAgent(context),'test_generator': TestGenerationAgent(context),'documenter': DocumentationAgent(context)}
defexecute_task(self, task):"""执行任务"""results ={}
# 步骤1: 生成代码print("Step 1: Generating code...")code_result = self.agents['code_generator'].generate_code(task)results['code']= code_result

# 步骤2: 审查代码print("Step 2: Reviewing code...")review_result = self.agents['code_reviewer'].review_code(code_result['code'],language=task.get('language','python'))results['review']= review_result

# 步骤3: 修复问题if review_result['summary']['critical_issues']>0:print("Step 3: Fixing critical issues...")fixed_code = self._fix_issues(code_result['code'],review_result['issues'])results['fixed_code']= fixed_code
else:results['fixed_code']= code_result['code']
# 步骤4: 生成测试print("Step 4: Generating tests...")test_result = self.agents['test_generator'].generate_tests(results['fixed_code'],language=task.get('language','python'))results['tests']= test_result

# 步骤5: 生成文档print("Step 5: Generating documentation...")doc_result = self.agents['documenter'].generate_documentation(results['fixed_code'],test_result['tests'])results['documentation']= doc_result

return results

def_fix_issues(self, code, issues):"""修复代码问题"""fixed_code = code

for issue in issues:if issue['type']=='security':fixed_code = self._fix_security_issue(fixed_code, issue)elif issue['type']=='high_complexity':fixed_code = self._refactor_complex_function(fixed_code, issue)
return fixed_code

def_fix_security_issue(self, code, issue):"""修复安全问题"""# 实现安全修复逻辑return code

def_refactor_complex_function(self, code, issue):"""重构复杂函数"""# 实现重构逻辑return code

```>**协作流程**:1. 代码生成子Agent生成代码
2. 代码审查子Agent审查代码
3. 修复发现的问题
4. 测试生成子Agent生成测试
5. 文档生成子Agent生成文档
```
### 2. 并行协作
# 示例：子Agent并行协作用户请求：
"使用子Agent并行处理多个任务"Claude Code 生成的并行协作流程：
````python
`python

import concurrent.futures

classParallelAgentOrchestrator:"""并行Agent协调器"""
def__init__(self, context):self.context = context
self.agents ={'code_generator': CodeGenerationAgent(context),'code_reviewer': CodeReviewAgent(context),'test_generator': TestGenerationAgent(context),'documenter': DocumentationAgent(context)}
defexecute_parallel_tasks(self, tasks):"""并行执行任务"""results ={}
with concurrent.futures.ThreadPoolExecutor(max_workers=4)as executor:# 提交任务future_to_task ={executor.submit(self.agents['code_generator'].generate_code, task): task
for task in tasks
}
# 收集结果for future in concurrent.futures.as_completed(future_to_task):task = future_to_task[future]try:result = future.result()results[task['id']]= result
except Exception as e:results[task['id']]={'error':str(e)}
return results

defexecute_parallel_analysis(self, code):"""并行分析代码"""results ={}
with concurrent.futures.ThreadPoolExecutor(max_workers=4)as executor:# 提交分析任务futures ={'static_analysis': executor.submit(self.agents['code_reviewer']._static_analysis,code,'python'),'pattern_analysis': executor.submit(self.agents['code_reviewer']._check_patterns,code,'python'),'security_analysis': executor.submit(self.agents['code_reviewer']._check_security,code,'python'),'metrics_calculation': executor.submit(self.agents['code_reviewer']._calculate_metrics,code,'python')}
# 收集结果for name, future in futures.items():try:results[name]= future.result()except Exception as e:results[name]={'error':str(e)}
return results

```>**并行协作优势**:>- 提高任务执行速度
>- 充分利用系统资源
>- 缩短总执行时间

```
## 子Agent的优势
### 1. 提高效率
| 任务类型 | 单Agent | 多子Agent | 效率提升 ||----------|---------|-----------|----------|| 代码生成 + 审查 |10分钟 |6分钟 |40% ↑ || 多文件处理 |15分钟 |5分钟 |67% ↑ || 复杂任务分解 |20分钟 |8分钟 |60% ↑ |
### 2. 提高质量
| 指标 | 单Agent | 多子Agent | 改善 ||------|---------|-----------|------|| 代码质量 |75%|90%|+20%|| 错误率 |15%|5%|67% ↓ || 测试覆盖率 |70%|85%|+21%|
### 3. 提高可维护性
-**模块化**: 每个子Agent独立维护
-**可扩展**: 容易添加新的子Agent
-**可复用**: 子Agent可以在不同任务中复用

## 总结
子Agent概述包括：

1.**子Agent的概念**: 什么是子Agent、子Agent vs 主Agent
2.**子Agent的类型**: 代码生成子Agent、代码审查子Agent
3.**子Agent的协作**: 串行协作、并行协作
4.**子Agent的优势**: 提高效率、提高质量、提高可维护性

通过子Agent机制，Claude Code可以更高效地处理复杂的编程任务。

在下一节中，我们将探讨异步子Agent任务。

```
```