代码助手开发
构建一个能够理解需求、编写代码、调试bug、编写测试的多Agent协作系统。
项目目标
输入：自然语言描述的需求
处理：多Agent协作，分工完成任务
输出：完整的代码 + 测试 + 文档技术架构
┌─────────────┐
│ 项目经理 │ 分析需求，分配任务
└──────┬──────┘
│
┌──────────┼──────────┐
↓ ↓ ↓
┌───────┐ ┌───────┐ ┌───────┐
│ 开发者 │ │ 测试员 │ │ 审查员 │
└───────┘ └───────┘ └───────┘
│ │ │
└──────────┼──────────┘
↓
┌─────────────┐
│ 最终代码 │
└─────────────┘完整代码
项目结构
code-assistant/
├── main.py # 主程序
├── agents.py # Agent定义
├── graph.py # 工作流图
└── requirements.txtagents.py
python
"""多Agent定义"""
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from typing import TypedDict
model = ChatOpenAI(model="gpt-4o", temperature=0)
class CodeProjectState(TypedDict):
"""项目状态"""
requirements: str # 原始需求
design: str # 设计文档
code: str # 代码
tests: str # 测试代码
review: str # 审查意见
final_code: str # 最终代码
status: str # 当前状态
def project_manager(state: CodeProjectState) -> dict:
"""项目经理Agent - 分析需求"""
prompt = ChatPromptTemplate.from_messages([
("system", """你是项目经理，负责分析需求并制定开发计划。
输出格式：
## 功能需求
- 需求1
- 需求2
## 技术方案
- 使用技术
- 架构设计
## 开发任务
1. 任务1
2. 任务2
"""),
("user", "需求：{requirements}")
])
chain = prompt | model
result = chain.invoke({"requirements": state["requirements"]})
return {"design": result.content, "status": "designed"}
def developer(state: CodeProjectState) -> dict:
"""开发Agent - 编写代码"""
prompt = ChatPromptTemplate.from_messages([
("system", """你是资深开发者，根据设计文档编写代码。
要求：
1. 代码规范、可读性强
2. 添加必要的注释
3. 处理异常情况
4. 只输出代码，不要解释
输出格式：
```python
# 代码内容"""), ("user", """设计文档：
请编写实现代码。""") ])
chain = prompt | model
result = chain.invoke({"design": state["design"]})
return {"code": result.content, "status": "coded"}
def tester(state: CodeProjectState) -> dict: """测试Agent - 编写测试""" prompt = ChatPromptTemplate.from_messages([ ("system", """你是测试工程师，为代码编写测试用例。
要求：
- 使用pytest框架
- 覆盖正常情况
- 覆盖边界情况
- 覆盖异常情况
输出格式：
python
# 测试代码"""), ("user", """代码：
请编写测试代码。""") ])
chain = prompt | model
result = chain.invoke({"code": state["code"]})
return {"tests": result.content, "status": "tested"}
def reviewer(state: CodeProjectState) -> dict: """审查Agent - 代码审查""" prompt = ChatPromptTemplate.from_messages([ ("system", """你是代码审查专家，审查代码质量。
审查要点：
- 代码规范
- 潜在bug
- 性能问题
- 安全隐患
输出格式：
审查结果
- 通过/需要修改
问题列表
- 问题1
- 问题2
改进建议
- 建议1
- 建议2 """), ("user", """代码：
测试：
请审查代码。""") ])
chain = prompt | model
result = chain.invoke({
"code": state["code"],
"tests": state["tests"]
})
return {"review": result.content, "status": "reviewed"}
def integrator(state: CodeProjectState) -> dict: """整合Agent - 整合最终代码""" prompt = ChatPromptTemplate.from_messages([ ("system", """整合代码和测试，根据审查意见优化。
输出最终完整代码，包含：
- 实现代码
- 测试代码
- 使用说明 """), ("user", """代码：
测试：
审查意见：
请输出最终代码包。""") ])
chain = prompt | model
result = chain.invoke({
"code": state["code"],
"tests": state["tests"],
"review": state["review"]
})
return {"final_code": result.content, "status": "completed"}
### graph.py
```python
"""工作流图"""
from langgraph.graph import StateGraph, END
from agents import (
CodeProjectState,
project_manager,
developer,
tester,
reviewer,
integrator
)
def build_workflow():
"""构建工作流"""
workflow = StateGraph(CodeProjectState)
# 添加节点
workflow.add_node("manager", project_manager)
workflow.add_node("developer", developer)
workflow.add_node("tester", tester)
workflow.add_node("reviewer", reviewer)
workflow.add_node("integrator", integrator)
# 定义边
workflow.set_entry_point("manager")
workflow.add_edge("manager", "developer")
workflow.add_edge("developer", "tester")
workflow.add_edge("tester", "reviewer")
workflow.add_edge("reviewer", "integrator")
workflow.add_edge("integrator", END)
return workflow.compile()
# 创建工作流
code_assistant = build_workflow()main.py
python
"""主程序"""
from graph import code_assistant
def main():
print("=" * 60)
print("AI代码助手 - 多Agent协作版")
print("=" * 60)
# 获取需求
requirements = input("\n请输入你的需求：\n").strip()
if not requirements:
print("需求不能为空")
return
# 执行工作流
print("\n开始处理...\n")
result = code_assistant.invoke({
"requirements": requirements,
"design": "",
"code": "",
"tests": "",
"review": "",
"final_code": "",
"status": "started"
})
# 输出结果
print("\n" + "=" * 60)
print("【设计文档】")
print("=" * 60)
print(result["design"])
print("\n" + "=" * 60)
print("【最终代码】")
print("=" * 60)
print(result["final_code"])
print("\n" + "=" * 60)
print(f"状态: {result['status']}")
print("=" * 60)
if __name__ == "__main__":
main()运行示例
============================================================
AI代码助手 - 多Agent协作版
============================================================
请输入你的需求：
实现一个用户登录系统，支持用户名密码登录和验证码验证
开始处理...
============================================================
【设计文档】
============================================================
## 功能需求
- 用户名密码登录
- 验证码生成和验证
- 登录状态管理
## 技术方案
- 使用Python + Flask
- session管理登录状态
- 验证码使用random生成
## 开发任务
1. 实现用户模型
2. 实现验证码生成
3. 实现登录接口
4. 编写测试用例
============================================================
【最终代码】
============================================================
# user_auth.py - 用户登录系统
from flask import Flask, request, session
import random
import string
app = Flask(__name__)
app.secret_key = 'your-secret-key'
class UserAuth:
"""用户认证系统"""
def __init__(self):
self.users = {} # 模拟数据库
def register(self, username: str, password: str) -> bool:
"""用户注册"""
if username in self.users:
return False
self.users[username] = password
return True
def verify_password(self, username: str, password: str) -> bool:
"""验证密码"""
return self.users.get(username) == password
class CaptchaService:
"""验证码服务"""
@staticmethod
def generate(length: int = 4) -> str:
"""生成验证码"""
return ''.join(random.choices(string.digits, k=length))
@staticmethod
def verify(input_code: str, actual_code: str) -> bool:
"""验证验证码"""
return input_code == actual_code
auth = UserAuth()
captcha_service = CaptchaService()
@app.route('/login', methods=['POST'])
def login():
"""登录接口"""
username = request.form.get('username')
password = request.form.get('password')
captcha = request.form.get('captcha')
# 验证码检查
if not captcha_service.verify(captcha, session.get('captcha')):
return {'success': False, 'message': '验证码错误'}
# 密码验证
if auth.verify_password(username, password):
session['user'] = username
return {'success': True, 'message': '登录成功'}
return {'success': False, 'message': '用户名或密码错误'}
@app.route('/captcha', methods=['GET'])
def get_captcha():
"""获取验证码"""
code = captcha_service.generate()
session['captcha'] = code
return {'captcha': code}
# test_user_auth.py - 测试代码
import pytest
from user_auth import UserAuth, CaptchaService
class TestUserAuth:
def test_register(self):
auth = UserAuth()
assert auth.register('test', '123456') == True
assert auth.register('test', '123456') == False
def test_verify_password(self):
auth = UserAuth()
auth.register('test', '123456')
assert auth.verify_password('test', '123456') == True
assert auth.verify_password('test', 'wrong') == False
class TestCaptchaService:
def test_generate(self):
code = CaptchaService.generate(4)
assert len(code) == 4
assert code.isdigit()
def test_verify(self):
assert CaptchaService.verify('1234', '1234') == True
assert CaptchaService.verify('1234', '5678') == False
## 使用说明
1. 安装依赖: pip install flask pytest
2. 运行服务: python user_auth.py
3. 运行测试: pytest test_user_auth.py
登录流程:
1. GET /captcha 获取验证码
2. POST /login 提交用户名、密码、验证码扩展方向
- 支持更多语言 - JavaScript、Go、Java
- 代码执行验证 - 实际运行代码检查
- 性能优化建议 - 分析代码复杂度
- 集成CI/CD - 自动化部署
小结
本项目实现了：
- 多Agent协作架构
- 流水线式任务处理
- 完整的开发流程
- 自动化代码生成
下一步
继续学习 自动化脚本开发，了解更复杂的任务规划。