"""
COT思维链效果对比演示 - 主程序
功能：
1. 加载配置和测试用例
2. 使用三种提示词策略分别调用模型
3. 对比展示输出结果
4. 保存结果到JSON文件
使用方法：
python main.py
环境要求：
- Python 3.8+
- dashscope
- python-dotenv
- tabulate
"""
import os
import sys
import json
import time
from datetime import datetime
from tabulate import tabulate
# 导入项目模块
from config import MODEL_CONFIG, OUTPUT_CONFIG, PATH_CONFIG
from model_client import DashscopeModel, create_model
from prompt_templates import (
NO_COT_PROMPT,
ZERO_SHOT_COT_PROMPT,
FEW_SHOT_COT_PROMPT,
PROMPT_TEMPLATES
)
from test_cases import TEST_CASES
def run_comparison(model, test_case):
"""
运行单个测试用例的三种策略对比
Args:
model: 模型实例
test_case: 测试用例字典
Returns:
dict: 包含三种策略结果的字典
"""
question = test_case["question"]
# 策略1：无思维链
print(" 📝 正在运行：无思维链...")
prompt1 = NO_COT_PROMPT.format(question=question)
result1 = model.call(prompt1)
# 策略2：零样本思维链
print(" 📝 正在运行：零样本思维链...")
prompt2 = ZERO_SHOT_COT_PROMPT.format(question=question)
result2 = model.call(prompt2)
# 策略3：少样本思维链
print(" 📝 正在运行：少样本思维链...")
prompt3 = FEW_SHOT_COT_PROMPT.format(question=question)
result3 = model.call(prompt3)
return {
"test_case": test_case,
"no_cot": result1,
"zero_shot_cot": result2,
"few_shot_cot": result3
}
def print_comparison(result):
"""
打印单个测试用例的对比结果
Args:
result: 对比结果字典
"""
test_case = result["test_case"]
# 打印标题
print(f"\n{'='*80}")
print(f"测试用例 #{test_case['id']} - {test_case['category']} ({test_case['difficulty']})")
print(f"问题：{test_case['question']}")
print(f"{'='*80}")
# 表格展示基本信息
table_data = [
["策略", "是否成功", "消耗Token"],
["无思维链", "✅" if result["no_cot"]["success"] else "❌",
result["no_cot"].get("tokens", "N/A")],
["零样本COT", "✅" if result["zero_shot_cot"]["success"] else "❌",
result["zero_shot_cot"].get("tokens", "N/A")],
["少样本COT", "✅" if result["few_shot_cot"]["success"] else "❌",
result["few_shot_cot"].get("tokens", "N/A")]
]
print(tabulate(table_data, headers="firstrow", tablefmt="grid"))
print()
# 详细输出对比
print("--- 无思维链输出 ---")
if result["no_cot"]["success"]:
print(result["no_cot"]["content"])
else:
print(f"❌ 错误：{result['no_cot'].get('error', '未知错误')}")
print("\n--- 零样本思维链输出 ---")
if result["zero_shot_cot"]["success"]:
print(result["zero_shot_cot"]["content"])
else:
print(f"❌ 错误：{result['zero_shot_cot'].get('error', '未知错误')}")
print("\n--- 少样本思维链输出 ---")
if result["few_shot_cot"]["success"]:
print(result["few_shot_cot"]["content"])
else:
print(f"❌ 错误：{result['few_shot_cot'].get('error', '未知错误')}")
# 显示预期答案
print(f"\n💡 预期答案：{test_case.get('expected_answer', '无')}")
print(f"💡 提示：{test_case.get('hint', '无')}")
print()
def print_summary(all_results):
"""
打印所有测试的汇总统计
Args:
all_results: 所有测试结果列表
"""
print("\n" + "="*80)
print("📊 测试汇总统计")
print("="*80)
# 统计成功次数
success_counts = {"no_cot": 0, "zero_shot_cot": 0, "few_shot_cot": 0}
token_counts = {"no_cot": 0, "zero_shot_cot": 0, "few_shot_cot": 0}
for result in all_results:
for strategy in ["no_cot", "zero_shot_cot", "few_shot_cot"]:
if result[strategy]["success"]:
success_counts[strategy] += 1
token_counts[strategy] += result[strategy].get("tokens", 0)
total = len(all_results)
# 打印统计表格
summary_data = [
["策略", "成功次数", "成功率", "平均Token消耗"],
["无思维链",
f"{success_counts['no_cot']}/{total}",
f"{success_counts['no_cot']/total*100:.1f}%",
f"{token_counts['no_cot']//max(success_counts['no_cot'], 1)}"],
["零样本COT",
f"{success_counts['zero_shot_cot']}/{total}",
f"{success_counts['zero_shot_cot']/total*100:.1f}%",
f"{token_counts['zero_shot_cot']//max(success_counts['zero_shot_cot'], 1)}"],
["少样本COT",
f"{success_counts['few_shot_cot']}/{total}",
f"{success_counts['few_shot_cot']/total*100:.1f}%",
f"{token_counts['few_shot_cot']//max(success_counts['few_shot_cot'], 1)}"]
]
print(tabulate(summary_data, headers="firstrow", tablefmt="grid"))
print()
def save_results(all_results):
"""
保存所有结果到JSON文件
Args:
all_results: 所有测试结果列表
Returns:
str: 保存的文件路径
"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"cot_comparison_{timestamp}.json"
filepath = os.path.join(PATH_CONFIG["results_dir"], filename)
# 确保目录存在
os.makedirs(PATH_CONFIG["results_dir"], exist_ok=True)
# 构建保存的数据结构
save_data = {
"metadata": {
"timestamp": timestamp,
"model": MODEL_CONFIG["model_name"],
"temperature": MODEL_CONFIG["temperature"],
"total_cases": len(all_results)
},
"results": all_results
}
# 写入文件
with open(filepath, "w", encoding="utf-8") as f:
json.dump(save_data, f, ensure_ascii=False, indent=2)
return filepath
def run_single_case(model, case_id):
"""
运行单个测试用例
Args:
model: 模型实例
case_id: 测试用例ID
"""
from test_cases import get_test_case
test_case = get_test_case(case_id)
if not test_case:
print(f"❌ 找不到测试用例 #{case_id}")
return None
print(f"\n正在运行测试用例 #{case_id}...")
result = run_comparison(model, test_case)
print_comparison(result)
return result
def main():
"""主程序入口"""
print("="*80)
print("🧠 COT思维链效果对比演示")
print("="*80)
print()
# 检查命令行参数
use_mock = "--mock" in sys.argv
single_case = None
for i, arg in enumerate(sys.argv):
if arg == "--case" and i + 1 < len(sys.argv):
try:
single_case = int(sys.argv[i + 1])
except ValueError:
print("❌ --case 参数需要是整数")
return
# 初始化模型
try:
model = create_model(use_mock=use_mock)
print(f"✅ 模型初始化成功")
print(f" 模型名称：{model.model_name}")
print(f" 温度参数：{model.temperature}")
print()
except Exception as e:
print(f"❌ 模型初始化失败：{e}")
return
# 确定要运行的测试用例
if single_case:
from test_cases import get_test_case
test_case = get_test_case(single_case)
if not test_case:
print(f"❌ 找不到测试用例 #{single_case}")
return
cases_to_run = [test_case]
print(f"📋 运行模式：单个测试用例 (#{single_case})")
else:
cases_to_run = TEST_CASES
print(f"📋 运行模式：全部测试用例 ({len(TEST_CASES)}个)")
print()
# 运行测试
all_results = []
start_time = time.time()
for i, test_case in enumerate(cases_to_run, 1):
print(f"\n[{i}/{len(cases_to_run)}] 正在运行测试用例 #{test_case['id']}...")
try:
result = run_comparison(model, test_case)
all_results.append(result)
print_comparison(result)
# 在测试用例之间添加短暂延迟，避免API限流
if i < len(cases_to_run):
time.sleep(1)
except Exception as e:
print(f"❌ 运行测试用例 #{test_case['id']} 时发生错误：{e}")
continue
elapsed_time = time.time() - start_time
# 打印汇总
if len(all_results) > 1:
print_summary(all_results)
# 保存结果
if OUTPUT_CONFIG["save_json"] and all_results:
try:
filepath = save_results(all_results)
print(f"💾 结果已保存到：{filepath}")
except Exception as e:
print(f"❌ 保存结果时发生错误：{e}")
# 结束信息
print("\n" + "="*80)
print("✅ 所有测试完成！")
print(f"⏱️ 总耗时：{elapsed_time:.1f}秒")
print()
print("📊 总结：")
print(" • 思维链能够显著提升大模型在复杂推理任务上的表现")
print(" • 少样本COT通常比零样本COT效果更好，因为提供了明确的格式示例")
print(" • 无思维链策略在简单问题上可能更快，但在复杂推理上容易出错")
print("="*80)
if __name__ == "__main__":
main()