481 lines
20 KiB
Python
481 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
稷下学宫 - 八仙ADK辩论系统 (Gemini 2.5 Flash版)
|
||
使用Google ADK和Gemini 2.5 Flash模型实现八仙辩论
|
||
"""
|
||
|
||
import os
|
||
import asyncio
|
||
import json
|
||
from datetime import datetime
|
||
from typing import Dict, List, Any, Optional
|
||
from google.adk import Agent, Runner
|
||
from google.adk.sessions import InMemorySessionService
|
||
from google.genai import types
|
||
|
||
# 加载.env文件
|
||
try:
|
||
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
except ImportError:
|
||
print("⚠️ 未安装python-dotenv，请运行: pip install python-dotenv")
|
||
pass
|
||
|
||
class BaXianADKDebateSystem:
|
||
"""八仙ADK辩论系统"""
|
||
|
||
def __init__(self):
|
||
self.model = "gemini-2.5-flash"
|
||
self.agents = {}
|
||
self.debate_history = []
|
||
self.current_round = 0
|
||
self.max_rounds = 3
|
||
|
||
# 八仙角色定义
|
||
self.baxian_profiles = {
|
||
"铁拐李": {
|
||
"personality": "八仙之首，形象为街头乞丐，代表社会底层。性格刚直不阿，善于逆向思维和批判分析。你从底层民众的角度看问题，敢于质疑权贵，为弱势群体发声。",
|
||
"speaking_style": "直言不讳，接地气，善用反问和草根智慧",
|
||
"expertise": "批判思维、民生洞察、社会底层视角"
|
||
},
|
||
"吕洞宾": {
|
||
"personality": "理性务实的学者型仙人，善于分析问题本质和长远影响。你注重逻辑推理，能够平衡各方观点，寻求最优解决方案。",
|
||
"speaking_style": "条理分明，深入浅出，善用类比和归纳",
|
||
"expertise": "战略分析、系统思维、决策优化"
|
||
},
|
||
"何仙姑": {
|
||
"personality": "八仙中唯一的女性，温柔智慧，善于从情感和人文角度思考问题。你关注社会影响和人文关怀，注重和谐与平衡。",
|
||
"speaking_style": "温和理性，富有同理心，善用情感共鸣",
|
||
"expertise": "人文关怀、社会影响、情感分析"
|
||
},
|
||
"蓝采和": {
|
||
"personality": "贫穷的街头歌者，自由奔放的艺术家气质。你代表精神富足但物质贫乏的群体，善于从艺术和美学角度看待问题，关注精神层面的价值。",
|
||
"speaking_style": "活泼生动，富有想象力，善用诗歌和民谣",
|
||
"expertise": "创新思维、艺术视角、精神追求、民间智慧"
|
||
},
|
||
"韩湘子": {
|
||
"personality": "年轻有为的技术专家，对新技术和趋势敏感。你善于从技术角度分析问题，关注实现可行性和技术细节。",
|
||
"speaking_style": "专业严谨，数据驱动，善用技术术语",
|
||
"expertise": "技术分析、趋势预测、可行性评估"
|
||
},
|
||
"曹国舅": {
|
||
"personality": "皇亲国戚，贵族出身，代表上层社会。你具有政治敏感性和大局观，善于从政策和制度角度分析问题，关注权力结构和利益平衡，维护既得利益群体。",
|
||
"speaking_style": "稳重大气，政治敏锐，善用历史典故和朝堂礼仪",
|
||
"expertise": "政策分析、制度设计、权力博弈、上层社会视角"
|
||
},
|
||
"张果老": {
|
||
"personality": "年长智慧的长者，经验丰富，善于从历史和哲学角度看问题。你能提供深刻的人生智慧和历史洞察。",
|
||
"speaking_style": "深沉睿智，引经据典，善用哲理思辨",
|
||
"expertise": "历史洞察、哲学思辨、人生智慧"
|
||
},
|
||
"钟离权": {
|
||
"personality": "汉钟离，出身将门富贵，军事战略家。你善于从战略和执行角度分析问题，注重实战经验和资源配置，关注执行力和结果导向。代表富贵阶层的视角。",
|
||
"speaking_style": "果断坚定，战略清晰，善用军事比喻和资源分析",
|
||
"expertise": "战略规划、执行管理、风险控制、资源配置"
|
||
}
|
||
}
|
||
|
||
def create_agents(self) -> bool:
|
||
"""创建八仙智能体"""
|
||
try:
|
||
for name, profile in self.baxian_profiles.items():
|
||
# 构建系统提示词
|
||
system_prompt = f"""
|
||
你是{name}，{profile['personality']}
|
||
|
||
你的说话风格：{profile['speaking_style']}
|
||
你的专业领域：{profile['expertise']}
|
||
|
||
在辩论中，请：
|
||
1. 保持你的角色特色和专业视角
|
||
2. 提供有价值的观点和分析
|
||
3. 与其他仙人进行建设性的讨论
|
||
4. 每次发言控制在200字以内
|
||
5. 语言要生动有趣，符合你的性格特点
|
||
"""
|
||
|
||
# 创建ADK智能体
|
||
agent = Agent(
|
||
name=name,
|
||
model=self.model,
|
||
instruction=system_prompt
|
||
)
|
||
|
||
self.agents[name] = agent
|
||
print(f"✅ 创建智能体: {name}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 创建智能体失败: {e}")
|
||
return False
|
||
|
||
async def conduct_debate(self, topic: str, rounds: int = 3) -> Dict[str, Any]:
|
||
"""进行八仙辩论"""
|
||
self.max_rounds = rounds
|
||
debate_id = f"baxian_debate_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
|
||
|
||
print(f"\n🏛️ 稷下学宫 - 八仙论道")
|
||
print(f"📋 辩论主题: {topic}")
|
||
print(f"🔄 辩论轮次: {rounds}")
|
||
print(f"🤖 使用模型: {self.model}")
|
||
print("=" * 80)
|
||
|
||
# 辩论开场
|
||
opening_context = f"""
|
||
今日稷下学宫八仙齐聚，共同探讨「{topic}」这一重要议题。
|
||
|
||
请各位仙人从自己的专业角度和人生阅历出发，分享真知灼见。
|
||
让我们通过思辨碰撞，共同寻求智慧的火花。
|
||
"""
|
||
|
||
self.debate_history.append({
|
||
"type": "opening",
|
||
"content": opening_context,
|
||
"timestamp": datetime.now().isoformat()
|
||
})
|
||
|
||
# 进行多轮辩论
|
||
for round_num in range(1, rounds + 1):
|
||
print(f"\n🎯 第{round_num}轮辩论")
|
||
print("-" * 60)
|
||
|
||
await self._conduct_round(topic, round_num)
|
||
|
||
# 轮次间隔
|
||
if round_num < rounds:
|
||
await asyncio.sleep(1)
|
||
|
||
# 辩论总结
|
||
await self._generate_summary(topic)
|
||
|
||
# 保存辩论记录
|
||
result = {
|
||
"debate_id": debate_id,
|
||
"topic": topic,
|
||
"model": self.model,
|
||
"rounds": rounds,
|
||
"participants": list(self.agents.keys()),
|
||
"debate_history": self.debate_history,
|
||
"timestamp": datetime.now().isoformat()
|
||
}
|
||
|
||
# 创建输出目录（相对于项目根目录）
|
||
project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
output_dir = os.path.join(project_root, "outputs", "debates")
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
# 保存到文件
|
||
filename = f"baxian_debate_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
|
||
filepath = os.path.join(output_dir, filename)
|
||
with open(filepath, 'w', encoding='utf-8') as f:
|
||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||
|
||
print(f"\n💾 辩论记录已保存: {filepath}")
|
||
|
||
return result
|
||
|
||
async def _conduct_round(self, topic: str, round_num: int):
|
||
"""进行单轮辩论"""
|
||
# 按照对立统一原则安排发言顺序
|
||
# 基于docs/baxian_debate_order_guide.md的分组原则
|
||
|
||
if round_num == 1:
|
||
# 第一轮：按对立组依次发言
|
||
speaker_order = [
|
||
"吕洞宾", # 乾/男
|
||
"何仙姑", # 坤/女
|
||
"张果老", # 老
|
||
"韩湘子", # 少
|
||
"钟离权", # 富（汉钟离）
|
||
"蓝采和", # 贫
|
||
"曹国舅", # 贵
|
||
"铁拐李" # 贱
|
||
]
|
||
else:
|
||
# 后续轮次：对立组交替发言，增强辩论张力
|
||
speaker_order = [
|
||
"铁拐李", "曹国舅", # 贵贱对立
|
||
"蓝采和", "钟离权", # 贫富对立
|
||
"韩湘子", "张果老", # 老少对立
|
||
"何仙姑", "吕洞宾" # 男女对立
|
||
]
|
||
|
||
for speaker_name in speaker_order:
|
||
if speaker_name in self.agents:
|
||
await self._agent_speak(speaker_name, topic, round_num)
|
||
await asyncio.sleep(0.5) # 短暂间隔
|
||
|
||
async def _agent_speak(self, speaker_name: str, topic: str, round_num: int):
|
||
"""智能体发言"""
|
||
agent = self.agents[speaker_name]
|
||
|
||
# 构建上下文
|
||
context = self._build_context(speaker_name, topic, round_num)
|
||
|
||
try:
|
||
# 创建会话服务（如果还没有）
|
||
if not hasattr(self, 'session_service'):
|
||
self.session_service = InMemorySessionService()
|
||
self.session = await self.session_service.create_session(
|
||
state={},
|
||
app_name="八仙论道系统",
|
||
user_id="debate_user"
|
||
)
|
||
|
||
# 创建Runner
|
||
runner = Runner(
|
||
app_name="八仙论道系统",
|
||
agent=agent,
|
||
session_service=self.session_service
|
||
)
|
||
|
||
# 构建消息内容
|
||
content = types.Content(role='user', parts=[types.Part(text=context)])
|
||
|
||
# 生成回应
|
||
response_stream = runner.run_async(
|
||
user_id=self.session.user_id,
|
||
session_id=self.session.id,
|
||
new_message=content
|
||
)
|
||
|
||
# 收集响应
|
||
response_parts = []
|
||
async for event in response_stream:
|
||
# 过滤ADK系统调试信息
|
||
event_str = str(event)
|
||
if ('Event from an unknown agent' in event_str or
|
||
'event id:' in event_str or
|
||
'API_KEY' in event_str):
|
||
continue
|
||
|
||
if hasattr(event, 'content') and event.content:
|
||
if hasattr(event.content, 'parts') and event.content.parts:
|
||
for part in event.content.parts:
|
||
if hasattr(part, 'text') and part.text and part.text.strip():
|
||
text_content = str(part.text).strip()
|
||
# 进一步过滤调试信息
|
||
if (not text_content.startswith('Event from') and
|
||
'API_KEY' not in text_content and
|
||
'event id:' not in text_content and
|
||
'unknown agent' not in text_content):
|
||
response_parts.append(text_content)
|
||
elif hasattr(event, 'text') and event.text:
|
||
text_content = str(event.text).strip()
|
||
if (not text_content.startswith('Event from') and
|
||
'API_KEY' not in text_content and
|
||
'event id:' not in text_content and
|
||
'unknown agent' not in text_content):
|
||
response_parts.append(text_content)
|
||
|
||
response = ''.join(response_parts).strip()
|
||
|
||
# 记录发言
|
||
speech_record = {
|
||
"type": "speech",
|
||
"round": round_num,
|
||
"speaker": speaker_name,
|
||
"content": response,
|
||
"timestamp": datetime.now().isoformat()
|
||
}
|
||
|
||
self.debate_history.append(speech_record)
|
||
|
||
# 显示发言
|
||
print(f"\n🗣️ {speaker_name}:")
|
||
print(f"{response}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ {speaker_name} 发言失败: {e}")
|
||
# 记录错误
|
||
error_record = {
|
||
"type": "error",
|
||
"round": round_num,
|
||
"speaker": speaker_name,
|
||
"error": str(e),
|
||
"timestamp": datetime.now().isoformat()
|
||
}
|
||
self.debate_history.append(error_record)
|
||
|
||
def _build_context(self, speaker_name: str, topic: str, round_num: int) -> str:
|
||
"""构建发言上下文"""
|
||
# 获取最近的发言历史
|
||
recent_speeches = []
|
||
for record in self.debate_history[-6:]: # 最近6条记录
|
||
if record["type"] == "speech" and record["speaker"] != speaker_name:
|
||
recent_speeches.append(f"{record['speaker']}: {record['content']}")
|
||
|
||
context = f"""
|
||
辩论主题：{topic}
|
||
当前轮次：第{round_num}轮
|
||
|
||
"""
|
||
|
||
if recent_speeches:
|
||
context += "最近的讨论：\n" + "\n".join(recent_speeches[-3:]) + "\n\n"
|
||
|
||
if round_num == 1:
|
||
context += "请从你的专业角度对这个主题发表观点，阐述你的立场和理由。"
|
||
else:
|
||
context += "请结合前面的讨论，进一步阐述你的观点，或对其他仙人的观点进行回应和补充。"
|
||
|
||
return context
|
||
|
||
async def _generate_summary(self, topic: str):
|
||
"""生成辩论总结"""
|
||
print(f"\n📝 辩论总结")
|
||
print("=" * 60)
|
||
|
||
# 统计各仙人发言次数
|
||
speech_count = {}
|
||
for record in self.debate_history:
|
||
if record["type"] == "speech":
|
||
speaker = record["speaker"]
|
||
speech_count[speaker] = speech_count.get(speaker, 0) + 1
|
||
|
||
print(f"\n📊 发言统计:")
|
||
for speaker, count in speech_count.items():
|
||
print(f" {speaker}: {count}次发言")
|
||
|
||
# 可以添加更多总结逻辑
|
||
summary_record = {
|
||
"type": "summary",
|
||
"topic": topic,
|
||
"speech_count": speech_count,
|
||
"total_speeches": sum(speech_count.values()),
|
||
"timestamp": datetime.now().isoformat()
|
||
}
|
||
|
||
self.debate_history.append(summary_record)
|
||
|
||
def check_api_key() -> bool:
|
||
"""检查API密钥"""
|
||
# 优先使用 GOOGLE_API_KEY，如果没有则使用 GEMINI_API_KEY
|
||
google_api_key = os.getenv('GOOGLE_API_KEY')
|
||
gemini_api_key = os.getenv('GEMINI_API_KEY')
|
||
|
||
if google_api_key and gemini_api_key:
|
||
print("⚠️ 检测到同时设置了 GOOGLE_API_KEY 和 GEMINI_API_KEY")
|
||
print("📝 建议：统一使用 GOOGLE_API_KEY，将移除 GEMINI_API_KEY")
|
||
# 使用 GOOGLE_API_KEY
|
||
api_key = google_api_key
|
||
print(f"✅ 使用 GOOGLE_API_KEY (长度: {len(api_key)} 字符)")
|
||
return True
|
||
elif google_api_key:
|
||
print(f"✅ 使用 GOOGLE_API_KEY (长度: {len(google_api_key)} 字符)")
|
||
return True
|
||
elif gemini_api_key:
|
||
print(f"✅ 使用 GEMINI_API_KEY (长度: {len(gemini_api_key)} 字符)")
|
||
# 设置 GOOGLE_API_KEY 为 GEMINI_API_KEY 的值
|
||
os.environ['GOOGLE_API_KEY'] = gemini_api_key
|
||
return True
|
||
else:
|
||
print("❌ 未找到 GOOGLE_API_KEY 或 GEMINI_API_KEY 环境变量")
|
||
print("请设置环境变量: export GOOGLE_API_KEY=your_api_key")
|
||
print("或使用: doppler run -- python examples/debates/baxian_adk_gemini_debate.py")
|
||
return False
|
||
|
||
def demo_mode():
|
||
"""演示模式 - 模拟辩论过程"""
|
||
print("🎭 演示模式：八仙论道模拟")
|
||
print("=" * 60)
|
||
|
||
topic = "人工智能对未来社会的影响"
|
||
print(f"📋 辩论主题: {topic}")
|
||
print("🔄 辩论轮次: 2")
|
||
print("🤖 模拟模式: 演示版本")
|
||
print("=" * 80)
|
||
|
||
# 模拟八仙发言
|
||
demo_speeches = {
|
||
"铁拐李": "人工智能虽然强大，但我们不能盲目崇拜。技术的发展必须以人为本，警惕其可能带来的风险和挑战。",
|
||
"吕洞宾": "从长远来看，AI将重塑社会结构。我们需要理性分析其影响，制定合适的发展策略，平衡效率与公平。",
|
||
"何仙姑": "技术进步应该服务于人类福祉。我们要关注AI对就业、教育的影响，确保技术发展不会加剧社会不平等。",
|
||
"蓝采和": "AI为艺术创作开辟了新天地！想象一下，人机协作能创造出多么奇妙的作品，这是前所未有的创新机遇。",
|
||
"韩湘子": "从技术角度看，AI的算力和算法正在指数级增长。我们需要关注数据安全、隐私保护等技术挑战。",
|
||
"曹国舅": "政策制定者必须未雨绸缪，建立完善的AI治理框架，平衡创新发展与风险管控的关系。",
|
||
"张果老": "纵观历史，每次技术革命都伴随着社会变迁。AI亦如此，关键在于如何引导其造福人类。",
|
||
"钟离权": "战略上要重视AI的军事应用，确保国家安全。同时要有执行力，将AI政策落到实处。"
|
||
}
|
||
|
||
print("\n🎯 第1轮辩论")
|
||
print("-" * 60)
|
||
|
||
for name, speech in demo_speeches.items():
|
||
print(f"\n🗣️ {name}:")
|
||
print(f"{speech}")
|
||
import time
|
||
time.sleep(1)
|
||
|
||
print("\n📝 辩论总结")
|
||
print("=" * 60)
|
||
print("📊 发言统计:")
|
||
for name in demo_speeches.keys():
|
||
print(f" {name}: 1次发言")
|
||
|
||
print("\n🎉 演示完成！")
|
||
print("💡 要体验完整的AI辩论功能，请配置真实的 GOOGLE_API_KEY")
|
||
|
||
async def main():
|
||
"""主函数"""
|
||
print("🏛️ 稷下学宫 - 八仙ADK辩论系统 (Gemini 2.5 Flash版)")
|
||
print("🤖 使用Google ADK + Gemini 2.5 Flash模型")
|
||
print("🎭 八仙齐聚，共论天下大事")
|
||
print("\n📝 注意：运行过程中可能出现ADK系统调试信息，这是正常现象")
|
||
print(" 包括'Event from an unknown agent'等信息，不影响辩论功能")
|
||
print()
|
||
|
||
# 检查API密钥
|
||
if not check_api_key():
|
||
print("⚠️ 未找到有效的 GOOGLE_API_KEY，启动演示模式")
|
||
print("💡 请设置环境变量以体验完整功能: export GOOGLE_API_KEY=your_api_key")
|
||
print("📝 获取API密钥: https://aistudio.google.com/app/apikey")
|
||
print()
|
||
|
||
# 演示模式 - 模拟辩论过程
|
||
demo_mode()
|
||
return
|
||
|
||
# 创建辩论系统
|
||
debate_system = BaXianADKDebateSystem()
|
||
|
||
# 创建智能体
|
||
if not debate_system.create_agents():
|
||
print("❌ 智能体创建失败，无法进行辩论")
|
||
return
|
||
|
||
# 辩论主题
|
||
topics = [
|
||
"人工智能对未来社会的影响",
|
||
"数字货币与传统金融的博弈",
|
||
"元宇宙技术的发展前景",
|
||
"可持续发展与经济增长的平衡",
|
||
"教育数字化转型的机遇与挑战"
|
||
]
|
||
|
||
# 选择主题（可以随机选择或让用户选择）
|
||
topic = topics[0] # 默认使用第一个主题
|
||
|
||
try:
|
||
# 开始辩论
|
||
result = await debate_system.conduct_debate(topic, rounds=2)
|
||
|
||
if result:
|
||
print(f"\n🎉 辩论成功完成！")
|
||
print(f"📁 辩论ID: {result['debate_id']}")
|
||
print(f"🎯 参与者: {', '.join(result['participants'])}")
|
||
print(f"📊 总发言数: {len([r for r in result['debate_history'] if r['type'] == 'speech'])}")
|
||
else:
|
||
print("❌ 辩论失败")
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n👋 用户中断，辩论结束")
|
||
except Exception as e:
|
||
print(f"❌ 辩论过程中发生错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main()) |