class CompetitorAnalysisWorkflow:
def __init__(self, llm_client):
self.llm = llm_client
def run(self, product_info):
# 步骤 1:信息整理
step1_result = self.execute_step(
"信息整理",
STEP1_PROMPT.format(raw_product_info=product_info)
)
# 步骤 2 和 3:并行执行
step2_result = self.execute_step(
"功能对比",
STEP2_PROMPT.format(step1_output=step1_result)
)
step3_result = self.execute_step(
"定价分析",
STEP3_PROMPT.format(step1_output=step1_result)
)
# 步骤 4:SWOT 分析
step4_result = self.execute_step(
"SWOT 分析",
STEP4_PROMPT.format(
step2_output=step2_result,
step3_output=step3_result
)
)
# 步骤 5:报告生成
step5_result = self.execute_step(
"报告生成",
STEP5_PROMPT.format(
step1_output=step1_result,
step2_output=step2_result,
step3_output=step3_result,
step4_output=step4_result
)
)
# 步骤 6:质量检查(带重试)
final_report = self.quality_check_with_retry(step5_result)
return final_report
def execute_step(self, step_name, prompt):
print(f"执行步骤: {step_name}")
response = self.llm.generate(prompt)
return response
def quality_check_with_retry(self, report, max_retries=2):
for attempt in range(max_retries + 1):
check_result = self.execute_step(
"质量检查",
STEP6_PROMPT.format(step5_output=report)
)
if check_result.get("approved", False):
return report
# 如果未通过,进行修正
report = self.revise_report(
report,
check_result.get("suggestions", [])
)
return report