734 lines
29 KiB
Python
734 lines
29 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, send_workflow_start
|
||
from app.models import RequirementAnalysis, TestCaseResult, CodeGenerationResult
|
||
|
||
# 初始化日志
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _parse_json(content: str, error_prefix: str) -> dict:
|
||
"""解析 LLM 返回的 JSON，解析失败时尝试正则兜底。"""
|
||
try:
|
||
return json.loads(content)
|
||
except json.JSONDecodeError:
|
||
import re
|
||
m = re.search(r'\{.*\}', content, re.DOTALL)
|
||
if m:
|
||
return json.loads(m.group())
|
||
raise ValueError(f"{error_prefix}: {content}")
|
||
|
||
|
||
class ClarifyAgent:
|
||
"""需求澄清 Agent —— 判断需求是否完整，并追问用户补充信息"""
|
||
|
||
def __init__(self):
|
||
self.settings = get_settings()
|
||
self.client = OpenAI(api_key=self.settings.api_key, base_url=self.settings.base_url)
|
||
|
||
def start(self, raw_requirement: str) -> dict:
|
||
"""
|
||
用户第一次提交需求时调用。
|
||
返回 {"ready": bool, "question": str, "clarified_requirement": str}
|
||
"""
|
||
prompt = f"""你是一个资深的产品经理助手，负责在正式分析需求之前确认需求的完整性。
|
||
|
||
用户提交的需求：
|
||
{raw_requirement}
|
||
|
||
请判断该需求是否足够清晰，可以直接开始产品分析。
|
||
|
||
返回 JSON：
|
||
{{
|
||
"ready": true 或 false,
|
||
"question": "如果 ready=false，给用户一个简洁的追问；如果 ready=true，此字段为空字符串",
|
||
"clarified_requirement": "整合后的完整需求描述（即使 ready=false 也要输出当前已有的描述）"
|
||
}}
|
||
|
||
判断标准：
|
||
- 需求描述清楚要做什么
|
||
- 有基本的场景或目标
|
||
- 不需要追问过多细节，够 PM 开始分析即可
|
||
|
||
返回 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=500,
|
||
response_format={"type": "json_object"}
|
||
)
|
||
return _parse_json(response.choices[0].message.content, "ClarifyAgent.start 解析失败")
|
||
|
||
def continue_clarify(self, clarify_history: list[dict], user_reply: str) -> dict:
|
||
"""
|
||
用户补充信息后继续澄清。
|
||
clarify_history 格式：[{"role": "assistant"|"user", "content": str}, ...]
|
||
返回同 start() 的格式。
|
||
"""
|
||
messages = [
|
||
{"role": "system", "content": "你是一个产品经理助手，负责判断需求完整性，输出必须是严格的 JSON 格式。"}
|
||
] + clarify_history + [
|
||
{"role": "user", "content": user_reply}
|
||
]
|
||
|
||
suffix = """\n\n根据以上对话，判断需求现在是否足够清晰。返回 JSON：
|
||
{{
|
||
"ready": true 或 false,
|
||
"question": "如果 ready=false，继续追问；如果 ready=true，此字段为空字符串",
|
||
"clarified_requirement": "整合所有对话后的完整需求描述"
|
||
}}
|
||
返回 ONLY JSON，不要有其他文字。"""
|
||
messages.append({"role": "user", "content": suffix})
|
||
|
||
response = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=messages,
|
||
temperature=0.2,
|
||
max_tokens=500,
|
||
response_format={"type": "json_object"}
|
||
)
|
||
return _parse_json(response.choices[0].message.content, "ClarifyAgent.continue_clarify 解析失败")
|
||
|
||
|
||
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 _build_prompt(self, requirement: str) -> str:
|
||
return f"""你是一个资深的产品经理。请根据以下简单的需求描述，进行深入分析和完善。
|
||
|
||
需求描述：{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内容，不要有其他文字。"""
|
||
|
||
def analyze_requirement(self, simple_requirement: str) -> RequirementAnalysis:
|
||
"""
|
||
分析和完善简单的需求描述
|
||
|
||
Args:
|
||
simple_requirement: 用户提供的简单需求描述
|
||
|
||
Returns:
|
||
RequirementAnalysis: 包含完善后的需求信息
|
||
"""
|
||
response = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "system", "content": "你是一个资深的产品经理，擅长需求分析和拆解，输出必须是严格的 JSON 格式。"},
|
||
{"role": "user", "content": self._build_prompt(simple_requirement)}
|
||
],
|
||
temperature=0.2,
|
||
max_tokens=2000,
|
||
response_format={"type": "json_object"}
|
||
)
|
||
result = _parse_json(response.choices[0].message.content, "PMAgent 解析失败")
|
||
return RequirementAnalysis(**result)
|
||
|
||
def stream_analyze(self, simple_requirement: str):
|
||
"""
|
||
流式版需求分析。yield 文本块(str)，最后 yield (None, RequirementAnalysis) 作为哨兵。
|
||
"""
|
||
stream = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "system", "content": "你是一个资深的产品经理，擅长需求分析和拆解，输出必须是严格的 JSON 格式。"},
|
||
{"role": "user", "content": self._build_prompt(simple_requirement)}
|
||
],
|
||
temperature=0.2,
|
||
max_tokens=2000,
|
||
response_format={"type": "json_object"},
|
||
stream=True,
|
||
)
|
||
full_text = ""
|
||
for chunk in stream:
|
||
delta = chunk.choices[0].delta.content or ""
|
||
if delta:
|
||
full_text += delta
|
||
yield delta
|
||
result = _parse_json(full_text, "PMAgent stream 解析失败")
|
||
yield (None, RequirementAnalysis(**result))
|
||
|
||
def stream_refine(self, previous: RequirementAnalysis, feedback: str):
|
||
"""
|
||
流式版需求分析修改。yield 文本块(str)，最后 yield (None, RequirementAnalysis) 作为哨兵。
|
||
"""
|
||
prompt = f"""你是一个资深的产品经理。以下是你之前输出的需求分析结果，用户对此有修改意见，请根据意见调整输出。
|
||
|
||
之前的需求分析：
|
||
{json.dumps(previous, ensure_ascii=False, indent=2)}
|
||
|
||
用户的修改意见：
|
||
{feedback}
|
||
|
||
请在原有基础上修改，保持 JSON 格式不变，返回完整的修改后结果，返回 ONLY JSON，不要有其他文字。"""
|
||
|
||
stream = 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"},
|
||
stream=True,
|
||
)
|
||
full_text = ""
|
||
for chunk in stream:
|
||
delta = chunk.choices[0].delta.content or ""
|
||
if delta:
|
||
full_text += delta
|
||
yield delta
|
||
result = _parse_json(full_text, "PMAgent stream_refine 解析失败")
|
||
yield (None, RequirementAnalysis(**result))
|
||
|
||
def refine(self, previous: RequirementAnalysis, feedback: str) -> RequirementAnalysis:
|
||
"""
|
||
根据用户反馈修改已有的需求分析结果。
|
||
|
||
Returns:
|
||
RequirementAnalysis: 修改后的需求分析结果
|
||
"""
|
||
prompt = f"""你是一个资深的产品经理。以下是你之前输出的需求分析结果，用户对此有修改意见，请根据意见调整输出。
|
||
|
||
之前的需求分析：
|
||
{json.dumps(previous, ensure_ascii=False, indent=2)}
|
||
|
||
用户的修改意见：
|
||
{feedback}
|
||
|
||
请在原有基础上修改，保持 JSON 格式不变，返回完整的修改后结果，返回 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"}
|
||
)
|
||
result = _parse_json(response.choices[0].message.content, "PMAgent.refine 解析失败")
|
||
return RequirementAnalysis(**result)
|
||
|
||
|
||
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 _build_prompt(self, requirement_analysis: RequirementAnalysis) -> str:
|
||
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"]}
|
||
"""
|
||
return f"""你是一个资深的Python QA工程师。根据以下需求信息，生成全面的Python测试用例和测试策略。所有测试用例必须基于Python语言，步骤和预期结果要符合Python的类型系统、异常机制和pytest测试框架。
|
||
|
||
{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. 测试用例要包含明确的步骤和预期结果，步骤和预期结果必须符合Python语言特性（不要出现Java或其他语言的描述）
|
||
6. 步骤和预期结果必须用自然语言描述，不得包含任何代码片段或代码块，不要出现 ```、assert、assertEquals 等代码语法
|
||
7. 测试策略不要出现pytest或者Python这种字眼，应该是针对需求的测试方法论和思路描述，覆盖计划要说明如何确保测试覆盖所有功能和边界情况
|
||
|
||
返回ONLY JSON内容，不要有其他文字。"""
|
||
|
||
def generate_test_cases(self, requirement_analysis: RequirementAnalysis) -> TestCaseResult:
|
||
"""
|
||
基于需求分析生成测试用例
|
||
|
||
Args:
|
||
requirement_analysis: PM Agent的分析结果
|
||
|
||
Returns:
|
||
TestCaseResult: 包含测试用例的结果
|
||
"""
|
||
response = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "system", "content": "你是一个资深的 Python QA 工程师，擅长为 Python 应用程序设计测试用例，所有测试步骤和预期结果必须基于 Python 语言特性（如 pytest、动态类型、异常机制等），输出必须是严格的 JSON 格式。"},
|
||
{"role": "user", "content": self._build_prompt(requirement_analysis)}
|
||
],
|
||
temperature=0.2,
|
||
max_tokens=3000,
|
||
response_format={"type": "json_object"}
|
||
)
|
||
result = _parse_json(response.choices[0].message.content, "QAAgent 解析失败")
|
||
return TestCaseResult(**result)
|
||
|
||
def stream_generate_test_cases(self, requirement_analysis: RequirementAnalysis):
|
||
"""
|
||
流式版测试用例生成。yield 文本块(str)，最后 yield (None, TestCaseResult) 作为哨兵。
|
||
"""
|
||
stream = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "system", "content": "你是一个资深的 Python QA 工程师，擅长为 Python 应用程序设计测试用例，输出必须是严格的 JSON 格式。"},
|
||
{"role": "user", "content": self._build_prompt(requirement_analysis)}
|
||
],
|
||
temperature=0.2,
|
||
max_tokens=3000,
|
||
response_format={"type": "json_object"},
|
||
stream=True,
|
||
)
|
||
full_text = ""
|
||
for chunk in stream:
|
||
delta = chunk.choices[0].delta.content or ""
|
||
if delta:
|
||
full_text += delta
|
||
yield delta
|
||
result = _parse_json(full_text, "QAAgent stream 解析失败")
|
||
yield (None, TestCaseResult(**result))
|
||
|
||
def stream_refine(self, previous: TestCaseResult, feedback: str):
|
||
"""
|
||
流式版测试用例修改。yield 文本块(str)，最后 yield (None, TestCaseResult) 作为哨兵。
|
||
"""
|
||
prompt = f"""你是一个资深的Python QA工程师。以下是你之前输出的测试用例，用户对此有修改意见，请根据意见调整输出。
|
||
|
||
之前的测试用例：
|
||
{json.dumps(previous, ensure_ascii=False, indent=2)}
|
||
|
||
用户的修改意见：
|
||
{feedback}
|
||
|
||
请在原有基础上修改，保持 JSON 格式不变，返回完整的修改后结果。
|
||
步骤和预期结果必须用自然语言描述，不得包含代码片段。
|
||
返回 ONLY JSON，不要有其他文字。"""
|
||
|
||
stream = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "system", "content": "你是一个资深的 Python QA 工程师，输出必须是严格的 JSON 格式。"},
|
||
{"role": "user", "content": prompt}
|
||
],
|
||
temperature=0.2,
|
||
max_tokens=3000,
|
||
response_format={"type": "json_object"},
|
||
stream=True,
|
||
)
|
||
full_text = ""
|
||
for chunk in stream:
|
||
delta = chunk.choices[0].delta.content or ""
|
||
if delta:
|
||
full_text += delta
|
||
yield delta
|
||
result = _parse_json(full_text, "QAAgent stream_refine 解析失败")
|
||
yield (None, TestCaseResult(**result))
|
||
|
||
def refine(self, previous: TestCaseResult, feedback: str) -> TestCaseResult:
|
||
"""
|
||
根据用户反馈修改已有的测试用例结果。
|
||
|
||
Args:
|
||
previous: 上一次的测试用例结果
|
||
feedback: 用户的修改意见
|
||
|
||
Returns:
|
||
TestCaseResult: 修改后的测试用例结果
|
||
"""
|
||
prompt = f"""你是一个资深的Python QA工程师。以下是你之前输出的测试用例，用户对此有修改意见，请根据意见调整输出。
|
||
|
||
之前的测试用例：
|
||
{json.dumps(previous, ensure_ascii=False, indent=2)}
|
||
|
||
用户的修改意见：
|
||
{feedback}
|
||
|
||
请在原有基础上修改，保持 JSON 格式不变，返回完整的修改后结果。
|
||
步骤和预期结果必须用自然语言描述，不得包含代码片段。
|
||
返回 ONLY JSON，不要有其他文字。"""
|
||
|
||
response = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "system", "content": "你是一个资深的 Python QA 工程师，输出必须是严格的 JSON 格式。"},
|
||
{"role": "user", "content": prompt}
|
||
],
|
||
temperature=0.2,
|
||
max_tokens=3000,
|
||
response_format={"type": "json_object"}
|
||
)
|
||
result = _parse_json(response.choices[0].message.content, "QAAgent.refine 解析失败")
|
||
return TestCaseResult(**result)
|
||
|
||
|
||
class DevAgent:
|
||
"""开发Agent - 生成Python代码和单元测试"""
|
||
|
||
def __init__(self):
|
||
self.settings = get_settings()
|
||
self.client = OpenAI(api_key=self.settings.api_key, base_url=self.settings.base_url)
|
||
|
||
def _build_prompt(self, requirement_analysis: RequirementAnalysis, test_cases: TestCaseResult) -> str:
|
||
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}
|
||
"""
|
||
return f"""你是一个资深的Python开发工程师。根据以下需求和测试用例，生成高质量的Python实现代码和单元测试代码。
|
||
|
||
{requirement_text}
|
||
|
||
{test_summary}
|
||
|
||
请返回JSON格式的结果（必须是有效的JSON格式）：
|
||
{{
|
||
"java_code": "完整的Python实现代码（包含主模块和必要的辅助类/函数）",
|
||
"unit_tests": "使用pytest的单元测试代码",
|
||
"implementation_notes": "实现说明和注意事项"
|
||
}}
|
||
|
||
Python代码要求：
|
||
1. 使用 Python 3.10+ 语法和特性（如 match/case、类型注解、dataclass、pathlib、f-string、列表推导等），代码风格遵循 PEP 8
|
||
2. 包含详细的中文代码注释，类和函数必须有中文 docstring
|
||
3. 包含异常处理
|
||
4. 支持所有的功能需求
|
||
5. 考虑非功能需求（性能、安全等）
|
||
|
||
单元测试要求：
|
||
1. 使用 pytest，充分利用 @pytest.mark.parametrize、pytest.raises、fixture 等特性
|
||
2. 为每个公共函数/方法生成测试
|
||
3. 包含正常情况、边缘情况和异常情况的测试
|
||
4. 使用有意义的测试函数名称（如 test_xxx_when_xxx_should_xxx）
|
||
5. 每个测试函数内用中文注释标注 Given / When / Then 三个阶段
|
||
6. 测试代码要清晰易读
|
||
7. 重要：业务代码文件保存名为 implementation.py，测试代码文件保存名为 test_implementation.py，测试文件必须使用 `from implementation import ...` 或 `import implementation` 导入业务代码
|
||
|
||
implementation_notes要求返回中文实现说明，内容要具体且有指导意义，不能只是简单的总结性描述，要包含对关键设计决策的解释和对复杂逻辑的说明。
|
||
|
||
返回ONLY JSON内容，不要有其他文字。"""
|
||
|
||
def generate_code(
|
||
self,
|
||
requirement_analysis: RequirementAnalysis,
|
||
test_cases: TestCaseResult
|
||
) -> CodeGenerationResult:
|
||
"""
|
||
生成Python实现代码和单元测试代码
|
||
|
||
Args:
|
||
requirement_analysis: PM Agent的分析结果
|
||
test_cases: QA Agent的测试用例
|
||
|
||
Returns:
|
||
CodeGenerationResult: 包含Python代码和单元测试代码
|
||
"""
|
||
response = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "system", "content": "你是一个资深的 Python 3.10+ 开发工程师，擅长使用 Python 类型注解、dataclass、pytest 等特性编写高质量代码和单元测试，所有代码注释用中文，输出必须是严格的 JSON 格式。"},
|
||
{"role": "user", "content": self._build_prompt(requirement_analysis, test_cases)}
|
||
],
|
||
temperature=0.2,
|
||
max_tokens=8192,
|
||
response_format={"type": "json_object"}
|
||
)
|
||
result = _parse_json(response.choices[0].message.content, "DevAgent 解析失败")
|
||
return CodeGenerationResult(**result)
|
||
|
||
def stream_generate_code(
|
||
self,
|
||
requirement_analysis: RequirementAnalysis,
|
||
test_cases: TestCaseResult,
|
||
):
|
||
"""
|
||
流式版代码生成。yield 文本块(str)，最后 yield (None, CodeGenerationResult) 作为哨兵。
|
||
"""
|
||
stream = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "system", "content": "你是一个资深的 Python 3.10+ 开发工程师，擅长使用 Python 类型注解、dataclass、pytest 等特性编写高质量代码，所有代码注释用中文，输出必须是严格的 JSON 格式。"},
|
||
{"role": "user", "content": self._build_prompt(requirement_analysis, test_cases)}
|
||
],
|
||
temperature=0.2,
|
||
max_tokens=8192,
|
||
response_format={"type": "json_object"},
|
||
stream=True,
|
||
)
|
||
full_text = ""
|
||
for chunk in stream:
|
||
delta = chunk.choices[0].delta.content or ""
|
||
if delta:
|
||
full_text += delta
|
||
yield delta
|
||
result = _parse_json(full_text, "DevAgent stream 解析失败")
|
||
yield (None, CodeGenerationResult(**result))
|
||
|
||
def refine(
|
||
self,
|
||
previous: CodeGenerationResult,
|
||
requirement_analysis: RequirementAnalysis,
|
||
test_cases: TestCaseResult,
|
||
feedback: str
|
||
) -> CodeGenerationResult:
|
||
"""
|
||
根据用户反馈修改已有的代码生成结果。
|
||
|
||
Args:
|
||
previous: 上一次的代码生成结果
|
||
requirement_analysis: PM Agent的分析结果（供参考）
|
||
test_cases: QA Agent的测试用例（供参考）
|
||
feedback: 用户的修改意见
|
||
|
||
Returns:
|
||
CodeGenerationResult: 修改后的代码生成结果
|
||
"""
|
||
refine_prompt = f"""以下是你之前生成的代码结果，用户对此有修改意见，请根据意见调整输出。
|
||
|
||
用户的修改意见：
|
||
{feedback}
|
||
|
||
请在原有代码基础上修改，保持 JSON 格式不变，返回完整的修改后结果。
|
||
继续遵守原有的 Python 3.10+、中文注释、pytest 等所有要求。
|
||
返回 ONLY JSON，不要有其他文字。"""
|
||
|
||
response = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "system", "content": "你是一个资深的 Python 3.10+ 开发工程师，所有代码注释用中文，输出必须是严格的 JSON 格式。"},
|
||
{"role": "user", "content": self._build_prompt(requirement_analysis, test_cases)},
|
||
{"role": "assistant", "content": json.dumps(previous, ensure_ascii=False)},
|
||
{"role": "user", "content": refine_prompt}
|
||
],
|
||
temperature=0.2,
|
||
max_tokens=8192,
|
||
response_format={"type": "json_object"}
|
||
)
|
||
result = _parse_json(response.choices[0].message.content, "DevAgent.refine 解析失败")
|
||
return CodeGenerationResult(**result)
|
||
|
||
|
||
class FixAgent:
|
||
"""自动修复Agent - 根据pytest失败信息修复Python代码"""
|
||
|
||
def __init__(self):
|
||
self.settings = get_settings()
|
||
self.client = OpenAI(api_key=self.settings.api_key, base_url=self.settings.base_url)
|
||
|
||
def fix(self, code_generation: CodeGenerationResult, test_output: str) -> CodeGenerationResult:
|
||
"""
|
||
根据pytest输出自动修复业务代码和/或测试代码。
|
||
|
||
Args:
|
||
code_generation: 之前 DevAgent 的产出
|
||
test_output: pytest 执行的完整输出
|
||
|
||
Returns:
|
||
CodeGenerationResult: 修复后的完整代码
|
||
"""
|
||
prompt = f"""以下Python代码在运行单元测试时出现了失败，请根据pytest失败信息修复代码。
|
||
|
||
当前业务代码（implementation.py）：
|
||
```python
|
||
{code_generation["java_code"]}
|
||
```
|
||
|
||
当前测试代码（test_implementation.py）：
|
||
```python
|
||
{code_generation["unit_tests"]}
|
||
```
|
||
|
||
pytest 执行输出：
|
||
{test_output[:4000]}
|
||
|
||
请分析失败原因并修复（业务代码和/或测试代码中的问题 都可以修复）。
|
||
层层修复要径直至递推所有测试权都能通过。
|
||
业务代码文件名为 implementation.py，测试文件名为 test_implementation.py，测试文件导入必须用 from implementation import ...。
|
||
|
||
返回 JSON 格式：
|
||
{{
|
||
"java_code": "修复后的完整Python业务代码",
|
||
"unit_tests": "修复后的完整Python测试代码",
|
||
"implementation_notes": "修复说明：具体说明了哪些问题，如何修复的"
|
||
}}
|
||
|
||
返回 ONLY JSON，不要有其他文字。"""
|
||
|
||
response = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "system", "content": "你是一个资深的 Python 3.10+ 开发工程师，擅长根据测试错误信息定位并修复 Bug，所有代码注释用中文，输出必须是严格的 JSON 格式。"},
|
||
{"role": "user", "content": prompt}
|
||
],
|
||
temperature=0.2,
|
||
max_tokens=8192,
|
||
response_format={"type": "json_object"}
|
||
)
|
||
result = _parse_json(response.choices[0].message.content, "FixAgent 解析失败")
|
||
return CodeGenerationResult(**result)
|
||
|
||
def stream_fix(self, code_generation: CodeGenerationResult, test_output: str):
|
||
"""
|
||
流式修复。yield 文本块(str)，最后 yield (None, CodeGenerationResult) 作为哨兵。
|
||
"""
|
||
prompt = f"""以下Python代码在运行单元测试时出现了失败，请根据pytest失败信息修复代码。
|
||
|
||
当前业务代码（implementation.py）：
|
||
```python
|
||
{code_generation["java_code"]}
|
||
```
|
||
|
||
当前测试代码（test_implementation.py）：
|
||
```python
|
||
{code_generation["unit_tests"]}
|
||
```
|
||
|
||
pytest 执行输出：
|
||
{test_output[:4000]}
|
||
|
||
请分析失败原因并修复（业务代码和/或测试代码中的问题 都可以修复）。
|
||
层层修复要径直至递推所有测试权都能通过。
|
||
业务代码文件名为 implementation.py，测试文件名为 test_implementation.py，测试文件导入必须用 from implementation import ...。
|
||
|
||
返回 JSON 格式：
|
||
{{
|
||
"java_code": "修复后的完整Python业务代码",
|
||
"unit_tests": "修复后的完整Python测试代码",
|
||
"implementation_notes": "修复说明：具体说明了哪些问题，如何修复的"
|
||
}}
|
||
|
||
返回 ONLY JSON，不要有其他文字。"""
|
||
|
||
stream = self.client.chat.completions.create(
|
||
model=self.settings.model,
|
||
messages=[
|
||
{"role": "system", "content": "你是一个资深的 Python 3.10+ 开发工程师，擅长根据测试错误信息定位并修复 Bug，所有代码注释用中文，输出必须是严格的 JSON 格式。"},
|
||
{"role": "user", "content": prompt}
|
||
],
|
||
temperature=0.2,
|
||
max_tokens=8192,
|
||
response_format={"type": "json_object"},
|
||
stream=True,
|
||
)
|
||
full_text = ""
|
||
for chunk in stream:
|
||
delta = chunk.choices[0].delta.content or ""
|
||
if delta:
|
||
full_text += delta
|
||
yield delta
|
||
result = _parse_json(full_text, "FixAgent stream 解析失败")
|
||
yield (None, CodeGenerationResult(**result))
|
||
|
||
|
||
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
|
||
}
|
||
|