demo2-qwen/app/agents.py

"""
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, send_workflow_start
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": "system", "content": "你是一个资深的产品经理，擅长需求分析和拆解，输出必须是严格的 JSON 格式。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.2,
            max_tokens=2000,
            response_format={"type": "json_object"}
        )

        # 提取响应内容
        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"""你是一个资深的Java QA工程师。根据以下需求信息，生成全面的Java测试用例和测试策略。所有测试用例必须基于Java语言，步骤和预期结果要符合Java的类型系统、异常机制和JUnit测试框架。

{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. 测试用例要包含明确的步骤和预期结果，步骤和预期结果必须符合Java语言特性（不要出现Python或其他语言的描述）
6. 步骤和预期结果必须用自然语言描述，不得包含任何代码片段或代码块，不要出现 ```、assert、assertEquals 等代码语法
7. 测试策略不要出现JUnit或者Java这种字眼，应该是针对需求的测试方法论和思路描述，覆盖计划要说明如何确保测试覆盖所有功能和边界情况

返回ONLY JSON内容，不要有其他文字。"""

        response = self.client.chat.completions.create(
            model=self.settings.model,
            messages=[
                {"role": "system", "content": "你是一个资深的 Java QA 工程师，擅长为 Java 应用程序设计测试用例，所有测试步骤和预期结果必须基于 Java 语言特性（如 JUnit、强类型系统、异常机制等），输出必须是严格的 JSON 格式。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.2,
            max_tokens=3000,
            response_format={"type": "json_object"}
        )

        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_cases_text = chr(10).join(
            f"- [{c['test_id']}] {c['test_name']}（类型：{c.get('test_type', '未分类')}）｜预期结果：{c['expected_result']}"
            for c in test_cases["test_cases"]
        )
        test_summary = f"""
测试用例数量: {len(test_cases["test_cases"])}
测试策略: {test_cases["test_strategy"]}
覆盖计划: {test_cases["coverage_plan"]}

关键测试用例列表：
{test_cases_text}
"""

        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 11 语法和特性（如 var 局部变量类型推断、String::isBlank/strip/lines、Optional、Stream API、List.of/Map.of 等不可变集合工厂方法），不要使用 Java 8 以前的写法
2. 包含详细的中文代码注释，所有注释内容必须用中文撰写，所有多行注释必须以 /* 开头、以 */ 结尾，Javadoc注释以 /** 开头、以 */ 结尾，绝对不能用单独的 / 作为注释结尾
3. 包含异常处理
4. 支持所有的功能需求
5. 考虑非功能需求（性能、安全等）

单元测试要求：
1. 使用 JUnit 5（jupiter），充分利用 @DisplayName、@ParameterizedTest、assertThrows 等特性
2. 为每个公共方法生成测试
3. 包含正常情况、边缘情况和异常情况的测试
4. 使用有意义的测试方法名称
5. 每个测试类顶部加中文类级注释说明该类的测试范围
6. 测试代码要清晰易读

implementation_notes要求返回中文实现说明，内容要具体且有指导意义，不能只是简单的总结性描述，要包含对关键设计决策的解释和对复杂逻辑的说明。

返回ONLY JSON内容，不要有其他文字。"""

        response = self.client.chat.completions.create(
            model=self.settings.model,
            messages=[
                {"role": "system", "content": "你是一个资深的 Java 11 开发工程师，擅长使用 Java 11 特性（var、Stream API、Optional、HttpClient、String 新方法等）编写高质量代码和单元测试，所有代码注释用中文，类名和方法名保持英文命名规范，输出必须是严格的 JSON 格式。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.2,
            max_tokens=8192,
            response_format={"type": "json_object"}
        )

        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_workflow_start(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
    }