258 lines
11 KiB
Python
258 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
八仙分层辩论系统：强模型分解 + 小模型辩论
|
||
架构：
|
||
1. 强模型（如GPT-4）进行问题分解和观点提炼
|
||
2. 小模型（如Gemini Flash）基于分解结果进行辩论
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import time
|
||
from typing import Dict, List, Any
|
||
import aiohttp
|
||
import logging
|
||
|
||
# 配置日志
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class BreakdownDebateSystem:
|
||
def __init__(self):
|
||
# API配置
|
||
self.api_base = "http://localhost:4000"
|
||
self.api_key = "sk-1234"
|
||
|
||
# 模型配置
|
||
self.strong_model = "fireworks_ai/accounts/fireworks/models/deepseek-v3-0324" # 强模型用于分解
|
||
self.debate_model = "gemini/gemini-2.5-flash" # 小模型用于辩论
|
||
|
||
# 辩论主题
|
||
self.topic = "工作量证明vs无限制爬虫：从李时珍采药到AI数据获取的激励机制变革"
|
||
|
||
# 八仙角色定义
|
||
self.immortals = {
|
||
"吕洞宾": {"性别": "男", "特征": "文雅学者，理性分析", "立场": "支持工作量证明机制"},
|
||
"何仙姑": {"性别": "女", "特征": "温和智慧，注重平衡", "立场": "支持无限制数据获取"},
|
||
"张果老": {"年龄": "老", "特征": "经验丰富，传统智慧", "立场": "支持传统激励机制"},
|
||
"韩湘子": {"年龄": "少", "特征": "创新思维，前瞻视野", "立场": "支持AI时代新机制"},
|
||
"汉钟离": {"地位": "富", "特征": "资源丰富，商业思维", "立场": "支持市场化激励"},
|
||
"蓝采和": {"地位": "贫", "特征": "平民视角，公平关注", "立场": "支持开放共享"},
|
||
"曹国舅": {"出身": "贵", "特征": "权威地位，规则意识", "立场": "支持制度化管理"},
|
||
"铁拐李": {"出身": "贱", "特征": "草根智慧，实用主义", "立场": "支持去中心化"}
|
||
}
|
||
|
||
# 对角线辩论配置
|
||
self.debate_pairs = [
|
||
("吕洞宾", "何仙姑"), # 男女对角线
|
||
("张果老", "韩湘子"), # 老少对角线
|
||
("汉钟离", "蓝采和"), # 富贫对角线
|
||
("曹国舅", "铁拐李") # 贵贱对角线
|
||
]
|
||
|
||
async def call_api(self, model: str, messages: List[Dict], max_tokens: int = 1000) -> str:
|
||
"""调用API"""
|
||
headers = {
|
||
"Authorization": f"Bearer {self.api_key}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
data = {
|
||
"model": model,
|
||
"messages": messages,
|
||
"max_tokens": max_tokens,
|
||
"temperature": 0.7
|
||
}
|
||
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(f"{self.api_base}/chat/completions",
|
||
headers=headers, json=data) as response:
|
||
if response.status == 200:
|
||
result = await response.json()
|
||
return result['choices'][0]['message']['content']
|
||
else:
|
||
error_text = await response.text()
|
||
logger.error(f"API调用失败: {response.status} - {error_text}")
|
||
return f"API调用失败: {response.status}"
|
||
except Exception as e:
|
||
logger.error(f"API调用异常: {str(e)}")
|
||
return f"API调用异常: {str(e)}"
|
||
|
||
async def breakdown_topic(self) -> Dict[str, Any]:
|
||
"""使用强模型分解辩论主题"""
|
||
logger.info("🧠 开始使用强模型分解辩论主题...")
|
||
|
||
breakdown_prompt = f"""
|
||
你是一个专业的辩论分析师。请对以下主题进行深度分解：
|
||
|
||
主题：{self.topic}
|
||
|
||
请提供：
|
||
1. 核心争议点（3-5个）
|
||
2. 支持工作量证明机制的关键论据（3个）
|
||
3. 支持无限制爬虫/数据获取的关键论据（3个）
|
||
4. 历史对比分析要点
|
||
5. 未来发展趋势预测
|
||
|
||
请以JSON格式返回，结构如下：
|
||
{{
|
||
"core_issues": ["争议点1", "争议点2", ...],
|
||
"pro_pow_arguments": ["论据1", "论据2", "论据3"],
|
||
"pro_unlimited_arguments": ["论据1", "论据2", "论据3"],
|
||
"historical_analysis": ["要点1", "要点2", ...],
|
||
"future_trends": ["趋势1", "趋势2", ...]
|
||
}}
|
||
"""
|
||
|
||
messages = [
|
||
{"role": "system", "content": "你是一个专业的辩论分析师，擅长深度分析复杂议题。"},
|
||
{"role": "user", "content": breakdown_prompt}
|
||
]
|
||
|
||
response = await self.call_api(self.strong_model, messages, max_tokens=2000)
|
||
|
||
try:
|
||
# 尝试解析JSON
|
||
breakdown_data = json.loads(response)
|
||
logger.info("✅ 主题分解完成")
|
||
return breakdown_data
|
||
except json.JSONDecodeError:
|
||
logger.error("❌ 强模型返回的不是有效JSON，使用默认分解")
|
||
return {
|
||
"core_issues": ["激励机制公平性", "创作者权益保护", "技术发展与伦理平衡"],
|
||
"pro_pow_arguments": ["保护创作者权益", "维护内容质量", "建立可持续生态"],
|
||
"pro_unlimited_arguments": ["促进知识传播", "加速技术发展", "降低获取成本"],
|
||
"historical_analysis": ["从手工采药到工业化生产的变迁", "知识产权制度的演进"],
|
||
"future_trends": ["AI与人类协作模式", "新型激励机制探索"]
|
||
}
|
||
|
||
async def conduct_debate(self, breakdown_data: Dict[str, Any]):
|
||
"""基于分解结果进行八仙辩论"""
|
||
logger.info("🎭 开始八仙对角线辩论...")
|
||
|
||
for i, (immortal1, immortal2) in enumerate(self.debate_pairs, 1):
|
||
logger.info(f"\n{'='*60}")
|
||
logger.info(f"第{i}场辩论：{immortal1} vs {immortal2}")
|
||
logger.info(f"{'='*60}")
|
||
|
||
# 为每个仙人准备个性化的论据
|
||
immortal1_info = self.immortals[immortal1]
|
||
immortal2_info = self.immortals[immortal2]
|
||
|
||
# 第一轮：开场陈述
|
||
statement1 = await self.get_opening_statement(immortal1, immortal1_info, breakdown_data)
|
||
logger.info(f"\n🗣️ {immortal1}的开场陈述：")
|
||
logger.info(statement1)
|
||
|
||
statement2 = await self.get_opening_statement(immortal2, immortal2_info, breakdown_data)
|
||
logger.info(f"\n🗣️ {immortal2}的开场陈述：")
|
||
logger.info(statement2)
|
||
|
||
# 第二轮：相互回应
|
||
response1 = await self.get_response(immortal1, immortal1_info, statement2, breakdown_data)
|
||
logger.info(f"\n💬 {immortal1}的回应：")
|
||
logger.info(response1)
|
||
|
||
response2 = await self.get_response(immortal2, immortal2_info, statement1, breakdown_data)
|
||
logger.info(f"\n💬 {immortal2}的回应：")
|
||
logger.info(response2)
|
||
|
||
# 第三轮：总结陈词
|
||
summary1 = await self.get_summary(immortal1, immortal1_info, [statement1, statement2, response1, response2], breakdown_data)
|
||
logger.info(f"\n📝 {immortal1}的总结：")
|
||
logger.info(summary1)
|
||
|
||
await asyncio.sleep(2) # 短暂停顿
|
||
|
||
logger.info(f"\n{'='*60}")
|
||
logger.info("🎉 所有四场对角线辩论已完成！")
|
||
logger.info(f"{'='*60}")
|
||
|
||
async def get_opening_statement(self, immortal: str, immortal_info: Dict, breakdown_data: Dict) -> str:
|
||
"""获取开场陈述"""
|
||
prompt = f"""
|
||
你是{immortal}，{immortal_info['特征']}。你的立场是：{immortal_info['立场']}。
|
||
|
||
基于以下分解分析，请发表你的开场陈述：
|
||
|
||
核心争议点：{', '.join(breakdown_data['core_issues'])}
|
||
|
||
支持工作量证明的论据：{', '.join(breakdown_data['pro_pow_arguments'])}
|
||
支持无限制获取的论据：{', '.join(breakdown_data['pro_unlimited_arguments'])}
|
||
|
||
历史分析要点：{', '.join(breakdown_data['historical_analysis'])}
|
||
未来趋势：{', '.join(breakdown_data['future_trends'])}
|
||
|
||
请以{immortal}的身份和特征，结合你的立场，发表一段150字左右的开场陈述。要体现你的个性特征和观点倾向。
|
||
"""
|
||
|
||
messages = [
|
||
{"role": "system", "content": f"你是{immortal}，请保持角色一致性。"},
|
||
{"role": "user", "content": prompt}
|
||
]
|
||
|
||
return await self.call_api(self.debate_model, messages)
|
||
|
||
async def get_response(self, immortal: str, immortal_info: Dict, opponent_statement: str, breakdown_data: Dict) -> str:
|
||
"""获取回应"""
|
||
prompt = f"""
|
||
你是{immortal}，{immortal_info['特征']}。你的立场是：{immortal_info['立场']}。
|
||
|
||
对方刚才说：
|
||
{opponent_statement}
|
||
|
||
基于分解分析的要点：
|
||
{', '.join(breakdown_data['core_issues'])}
|
||
|
||
请以{immortal}的身份回应对方的观点，约100字。要体现你的立场和特征。
|
||
"""
|
||
|
||
messages = [
|
||
{"role": "system", "content": f"你是{immortal}，请保持角色一致性。"},
|
||
{"role": "user", "content": prompt}
|
||
]
|
||
|
||
return await self.call_api(self.debate_model, messages)
|
||
|
||
async def get_summary(self, immortal: str, immortal_info: Dict, all_statements: List[str], breakdown_data: Dict) -> str:
|
||
"""获取总结陈词"""
|
||
prompt = f"""
|
||
你是{immortal}，{immortal_info['特征']}。你的立场是：{immortal_info['立场']}。
|
||
|
||
基于刚才的辩论内容和分解分析，请发表你的总结陈词，约120字。
|
||
要总结你的核心观点，并展望未来。
|
||
|
||
分析要点：{', '.join(breakdown_data['future_trends'])}
|
||
"""
|
||
|
||
messages = [
|
||
{"role": "system", "content": f"你是{immortal}，请保持角色一致性。"},
|
||
{"role": "user", "content": prompt}
|
||
]
|
||
|
||
return await self.call_api(self.debate_model, messages)
|
||
|
||
async def run(self):
|
||
"""运行完整的分层辩论系统"""
|
||
logger.info("🚀 启动八仙分层辩论系统")
|
||
logger.info(f"主题：{self.topic}")
|
||
logger.info(f"强模型（分解）：{self.strong_model}")
|
||
logger.info(f"辩论模型：{self.debate_model}")
|
||
|
||
# 第一阶段：强模型分解
|
||
breakdown_data = await self.breakdown_topic()
|
||
|
||
logger.info("\n📊 分解结果：")
|
||
for key, value in breakdown_data.items():
|
||
logger.info(f"{key}: {value}")
|
||
|
||
# 第二阶段：小模型辩论
|
||
await self.conduct_debate(breakdown_data)
|
||
|
||
logger.info("\n🎊 分层辩论系统运行完成！")
|
||
|
||
if __name__ == "__main__":
|
||
system = BreakdownDebateSystem()
|
||
asyncio.run(system.run()) |