306 lines
9.4 KiB
Python
306 lines
9.4 KiB
Python
"""
|
||
app/agents.py - 多Agent系统：PM Agent -> QA Agent -> Dev Agent 的工作流
|
||
"""
|
||
|
||
import logging
|
||
from app.config import get_settings
|
||
import json
|
||
from openai import OpenAI
|
||
|
||
from app.message import send_requirement_result, send_test_cases, send_message_to_feishu, send_generate_code
|
||
from app.models import RequirementAnalysis, TestCaseResult, CodeGenerationResult
|
||
|
||
# 初始化日志
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class PMAgent:
|
||
"""产品经理Agent - 完善和扩展需求"""
|
||
|
||
def __init__(self):
|
||
self.settings = get_settings()
|
||
self.client = OpenAI(api_key=self.settings.api_key, base_url=self.settings.base_url)
|
||
|
||
def analyze_requirement(self, simple_requirement: str) -> RequirementAnalysis:
|
||
"""
|
||
分析和完善简单的需求描述
|
||
|
||
Args:
|
||
simple_requirement: 用户提供的简单需求描述
|
||
|
||
Returns:
|
||
RequirementAnalysis: 包含完善后的需求信息
|
||
"""
|
||
|
||
prompt = f"""你是一个资深的产品经理。请根据以下简单的需求描述，进行深入分析和完善。
|
||
|
||
需求描述：{simple_requirement}
|
||
|
||
请按以下格式返回JSON结果（必须是有效的JSON格式）：
|
||
{{
|
||
"functional_requirements": ["功能需求1", "功能需求2", ...],
|
||
"non_functional_requirements": ["非功能需求1", "非功能需求2", ...],
|
||
"acceptance_criteria": ["验收标准1", "验收标准2", ...],
|
||
"edge_cases": ["边缘情况1", "边缘情况2", ...],
|
||
"summary": "需求总结说明"
|
||
}}
|
||
|
||
要求：
|
||
1. 识别至少3个功能需求
|
||
2. 识别至少2个非功能需求（性能、安全、可维护性等）
|
||
3. 为每个功能列出至少2个严格的验收标准
|
||
4. 识别至少3个潜在的边缘情况
|
||
5. 提供清晰的需求总结
|
||
|
||
返回ONLY JSON内容，不要有其他文字。"""
|
||
|
||
response = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "user", "content": prompt}
|
||
],
|
||
temperature=0.7,
|
||
max_tokens=2000
|
||
)
|
||
|
||
# 提取响应内容
|
||
content = response.choices[0].message.content
|
||
|
||
# 解析JSON
|
||
try:
|
||
result = json.loads(content)
|
||
return RequirementAnalysis(**result)
|
||
except json.JSONDecodeError:
|
||
# 如果JSON解析失败，尝试提取JSON部分
|
||
import re
|
||
json_match = re.search(r'\{.*\}', content, re.DOTALL)
|
||
if json_match:
|
||
result = json.loads(json_match.group())
|
||
return RequirementAnalysis(**result)
|
||
raise ValueError(f"无法解析Agent响应: {content}")
|
||
|
||
|
||
class QAAgent:
|
||
"""质量保证Agent - 生成测试用例"""
|
||
|
||
def __init__(self):
|
||
self.settings = get_settings()
|
||
self.client = OpenAI(api_key=self.settings.api_key, base_url=self.settings.base_url)
|
||
|
||
def generate_test_cases(self, requirement_analysis: RequirementAnalysis) -> TestCaseResult:
|
||
"""
|
||
基于需求分析生成测试用例
|
||
|
||
Args:
|
||
requirement_analysis: PM Agent的分析结果
|
||
|
||
Returns:
|
||
TestCaseResult: 包含测试用例的结果
|
||
"""
|
||
|
||
# 构建需求信息
|
||
requirement_text = f"""
|
||
功能需求：
|
||
{chr(10).join(f"- {req}" for req in requirement_analysis["functional_requirements"])}
|
||
|
||
非功能需求：
|
||
{chr(10).join(f"- {req}" for req in requirement_analysis["non_functional_requirements"])}
|
||
|
||
验收标准：
|
||
{chr(10).join(f"- {ac}" for ac in requirement_analysis["acceptance_criteria"])}
|
||
|
||
边缘情况：
|
||
{chr(10).join(f"- {ec}" for ec in requirement_analysis["edge_cases"])}
|
||
|
||
需求总结：
|
||
{requirement_analysis["summary"]}
|
||
"""
|
||
|
||
prompt = f"""你是一个资深的QA工程师。根据以下需求信息，生成全面的测试用例和测试策略。
|
||
|
||
{requirement_text}
|
||
|
||
请返回JSON格式的结果（必须是有效的JSON格式）：
|
||
{{
|
||
"test_cases": [
|
||
{{
|
||
"test_id": "TC001",
|
||
"test_name": "测试用例名称",
|
||
"precondition": "前置条件",
|
||
"steps": ["步骤1", "步骤2", ...],
|
||
"expected_result": "预期结果",
|
||
"test_type": "功能测试|性能测试|安全测试"
|
||
}},
|
||
...
|
||
],
|
||
"test_strategy": "测试策略说明",
|
||
"coverage_plan": "测试覆盖计划"
|
||
}}
|
||
|
||
要求：
|
||
1. 为每个功能需求生成至少2个测试用例
|
||
2. 为每个边缘情况生成1个测试用例
|
||
3. 生成至少1个性能测试用例
|
||
4. 生成至少1个安全测试用例
|
||
5. 测试用例要包含明确的步骤和预期结果
|
||
|
||
返回ONLY JSON内容，不要有其他文字。"""
|
||
|
||
response = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "user", "content": prompt}
|
||
],
|
||
temperature=0.7,
|
||
max_tokens=3000
|
||
)
|
||
|
||
content = response.choices[0].message.content
|
||
|
||
try:
|
||
result = json.loads(content)
|
||
return TestCaseResult(**result)
|
||
except json.JSONDecodeError:
|
||
import re
|
||
json_match = re.search(r'\{.*\}', content, re.DOTALL)
|
||
if json_match:
|
||
result = json.loads(json_match.group())
|
||
return TestCaseResult(**result)
|
||
raise ValueError(f"无法解析QA Agent响应: {content}")
|
||
|
||
|
||
class DevAgent:
|
||
"""开发Agent - 生成Java代码和单元测试"""
|
||
|
||
def __init__(self):
|
||
self.settings = get_settings()
|
||
self.client = OpenAI(api_key=self.settings.api_key, base_url=self.settings.base_url)
|
||
|
||
def generate_code(
|
||
self,
|
||
requirement_analysis: RequirementAnalysis,
|
||
test_cases: TestCaseResult
|
||
) -> CodeGenerationResult:
|
||
"""
|
||
生成Java实现代码和单元测试代码
|
||
|
||
Args:
|
||
requirement_analysis: PM Agent的分析结果
|
||
test_cases: QA Agent的测试用例
|
||
|
||
Returns:
|
||
CodeGenerationResult: 包含Java代码和单元测试代码
|
||
"""
|
||
|
||
# 构建上下文
|
||
requirement_text = f"""
|
||
功能需求：
|
||
{chr(10).join(f"- {req}" for req in requirement_analysis["functional_requirements"])}
|
||
|
||
非功能需求：
|
||
{chr(10).join(f"- {req}" for req in requirement_analysis["non_functional_requirements"])}
|
||
|
||
验收标准：
|
||
{chr(10).join(f"- {ac}" for ac in requirement_analysis["acceptance_criteria"])}
|
||
|
||
需求总结：
|
||
{requirement_analysis["summary"]}
|
||
"""
|
||
|
||
test_summary = f"""
|
||
测试用例数量: {len(test_cases["test_cases"])}
|
||
测试策略: {test_cases["test_strategy"]}
|
||
覆盖计划: {test_cases["coverage_plan"]}
|
||
"""
|
||
|
||
prompt = f"""你是一个资深的Java开发工程师。根据以下需求和测试用例，生成高质量的Java实现代码和单元测试代码。
|
||
|
||
{requirement_text}
|
||
|
||
{test_summary}
|
||
|
||
请返回JSON格式的结果（必须是有效的JSON格式）：
|
||
{{
|
||
"java_code": "完整的Java实现代码（包含主类和必要的辅助类）",
|
||
"unit_tests": "使用JUnit的单元测试代码",
|
||
"implementation_notes": "实现说明和注意事项",
|
||
"unit_tests_count": "生成的单元测试数量",
|
||
"passed_tests_count": "通过的单元测试数量"
|
||
}}
|
||
|
||
Java代码要求：
|
||
1. 遵循Java编码规范和最佳实践
|
||
2. 包含详细的代码注释
|
||
3. 包含异常处理
|
||
4. 支持所有的功能需求
|
||
5. 考虑非功能需求（性能、安全等）
|
||
|
||
单元测试要求：
|
||
1. 使用JUnit 4 或 JUnit 5
|
||
2. 为每个公共方法生成测试
|
||
3. 包含正常情况、边缘情况和异常情况的测试
|
||
4. 使用有意义的测试方法名称
|
||
5. 测试代码要清晰易读
|
||
|
||
通过的单元测试数量一定要是你真实运行了单元测试之后得到的真实数字，不能伪造
|
||
|
||
返回ONLY JSON内容，不要有其他文字。"""
|
||
|
||
response = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "user", "content": prompt}
|
||
],
|
||
temperature=0.7,
|
||
max_tokens=8192
|
||
)
|
||
|
||
content = response.choices[0].message.content
|
||
|
||
|
||
try:
|
||
result = json.loads(content)
|
||
return CodeGenerationResult(**result)
|
||
except json.JSONDecodeError:
|
||
import re
|
||
json_match = re.search(r'\{.*\}', content, re.DOTALL)
|
||
if json_match:
|
||
result = json.loads(json_match.group())
|
||
return CodeGenerationResult(**result)
|
||
raise ValueError(f"无法解析Dev Agent响应: {content}")
|
||
|
||
|
||
async def orchestrate_agents(simple_requirement: str) -> dict:
|
||
"""
|
||
编排三个Agent的工作流程
|
||
|
||
Args:
|
||
simple_requirement: 用户的简单需求描述
|
||
|
||
Returns:
|
||
包含所有Agent结果的完整字典
|
||
"""
|
||
|
||
# send_message_to_feishu(f"收到新需求: {simple_requirement}")
|
||
# Step 1: PM Agent 分析需求
|
||
pm_agent = PMAgent()
|
||
requirement_analysis = pm_agent.analyze_requirement(simple_requirement)
|
||
# send_requirement_result(requirement_analysis)
|
||
|
||
# Step 2: QA Agent 生成测试用例
|
||
qa_agent = QAAgent()
|
||
test_cases = qa_agent.generate_test_cases(requirement_analysis)
|
||
# send_test_cases(test_cases)
|
||
|
||
# Step 3: Dev Agent 生成代码
|
||
dev_agent = DevAgent()
|
||
code_generation = dev_agent.generate_code(requirement_analysis, test_cases)
|
||
# send_generate_code(code_generation)
|
||
return {
|
||
"requirement_analysis": requirement_analysis,
|
||
"test_cases": test_cases,
|
||
"code_generation": code_generation
|
||
}
|
||
|