#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 八仙辩论系统 - 自定义API版本 使用自定义LiteLLM端点而不是Google ADK """ import asyncio import aiohttp import json import os from typing import List, Dict, Any import time class CustomAPIAgent: """使用自定义API的代理""" def __init__(self, name: str, personality: str, api_url: str, api_key: str, model: str = "fireworks_ai/accounts/fireworks/models/deepseek-v3-0324"): self.name = name self.personality = personality self.api_url = api_url self.api_key = api_key self.model = model async def generate_response(self, prompt: str, session: aiohttp.ClientSession) -> str: """生成AI回应""" try: headers = { "Content-Type": "application/json", "x-litellm-api-key": self.api_key } payload = { "model": self.model, "messages": [ {"role": "system", "content": f"你是{self.name}，{self.personality}"}, {"role": "user", "content": prompt} ], "max_tokens": 1000, "temperature": 0.8 } async with session.post( f"{self.api_url}/v1/chat/completions", headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=30) ) as response: if response.status == 200: result = await response.json() content = result.get('choices', [{}])[0].get('message', {}).get('content', '') if content: return content.strip() else: print(f"❌ {self.name} API返回空内容: {result}") return f"[{self.name}暂时无法回应，API返回空内容]" else: error_text = await response.text() print(f"❌ {self.name} API错误 ({response.status}): {error_text[:200]}...") return f"[{self.name}暂时无法回应，API错误: {response.status}]" except Exception as e: print(f"❌ {self.name} 生成回应时出错: {e}") return f"[{self.name}暂时无法回应，连接错误]" class BaXianCustomDebateSystem: """八仙自定义API辩论系统""" def __init__(self, api_url: str, api_key: str): self.api_url = api_url.rstrip('/') self.api_key = api_key # 创建八仙代理 self.agents = { "吕洞宾": CustomAPIAgent( "吕洞宾", "八仙之首，男性代表，理性务实，善于分析问题的本质和长远影响。你代表男性视角，注重逻辑和实用性。", api_url, api_key ), "何仙姑": CustomAPIAgent( "何仙姑", "八仙中唯一的女性，温柔智慧，善于从情感和人文角度思考问题。你代表女性视角，注重关怀和和谐。", api_url, api_key ), "张果老": CustomAPIAgent( "张果老", "八仙中的长者，经验丰富，代表传统智慧和保守观点。你重视稳定和传承，谨慎对待变化。", api_url, api_key ), "韩湘子": CustomAPIAgent( "韩湘子", "八仙中的年轻人，充满活力和创新精神。你代表新生代观点，勇于尝试和改变。", api_url, api_key ), "汉钟离": CustomAPIAgent( "汉钟离", "八仙中的富贵者，见多识广，代表富裕阶层的观点。你注重效率和成果，善于资源配置。", api_url, api_key ), "蓝采和": CustomAPIAgent( "蓝采和", "八仙中的贫苦出身，朴实无华，代表普通民众的观点。你关注基层需求，重视公平正义。", api_url, api_key ), "曹国舅": CustomAPIAgent( "曹国舅", "八仙中的贵族，出身高贵，代表上层社会观点。你注重秩序和礼仪，维护既有体系。", api_url, api_key ), "铁拐李": CustomAPIAgent( "铁拐李", "八仙中的平民英雄，不拘小节，代表底层民众观点。你直言不讳，为弱势群体发声。", api_url, api_key ) } # 定义四对矛盾的对角线辩论 self.debate_pairs = [ ("吕洞宾", "何仙姑", "男女对立辩论"), ("张果老", "韩湘子", "老少对立辩论"), ("汉钟离", "蓝采和", "富贫对立辩论"), ("曹国舅", "铁拐李", "贵贱对立辩论") ] async def test_api_connection(self) -> bool: """测试API连接""" print(f"🔍 测试API连接: {self.api_url}") try: async with aiohttp.ClientSession() as session: headers = {"x-litellm-api-key": self.api_key} async with session.get( f"{self.api_url}/v1/models", headers=headers, timeout=aiohttp.ClientTimeout(total=10) ) as response: if response.status == 200: models = await response.json() print(f"✅ API连接成功，找到 {len(models.get('data', []))} 个模型") return True else: error_text = await response.text() print(f"❌ API连接失败 ({response.status}): {error_text[:200]}...") return False except Exception as e: print(f"❌ API连接测试失败: {e}") return False async def conduct_debate(self, topic: str) -> None: """进行完整的八仙辩论""" print(f"\n{'='*80}") print(f"🎭 八仙自定义API辩论系统") print(f"📝 辩论主题: {topic}") print(f"🔗 API端点: {self.api_url}") print(f"{'='*80}\n") # 测试API连接 if not await self.test_api_connection(): print("❌ API连接失败，无法进行辩论") return async with aiohttp.ClientSession() as session: for i, (agent1_name, agent2_name, debate_type) in enumerate(self.debate_pairs, 1): print(f"\n🎯 第{i}场辩论: {debate_type}") print(f"⚔️ {agent1_name} VS {agent2_name}") print(f"📋 主题: {topic}") print("-" * 60) agent1 = self.agents[agent1_name] agent2 = self.agents[agent2_name] # 第一轮：agent1开场 prompt1 = f"针对'{topic}'这个话题，请从你的角度阐述观点。要求：1)明确表达立场 2)提供具体论据 3)字数控制在200字以内" print(f"\n🗣️ {agent1_name}发言:") agent1_reply = await agent1.generate_response(prompt1, session) print(f"{agent1_reply}\n") # 第二轮：agent2回应 prompt2 = f"针对'{topic}'这个话题，{agent1_name}刚才说：'{agent1_reply}'。请从你的角度回应并阐述不同观点。要求：1)回应对方观点 2)提出自己的立场 3)字数控制在200字以内" print(f"🗣️ {agent2_name}回应:") agent2_reply = await agent2.generate_response(prompt2, session) print(f"{agent2_reply}\n") # 第三轮：agent1总结 prompt3 = f"针对'{topic}'这个话题的辩论，{agent2_name}回应说：'{agent2_reply}'。请做最后总结发言。要求：1)回应对方观点 2)强化自己立场 3)寻求共识或妥协 4)字数控制在150字以内" print(f"🗣️ {agent1_name}总结:") agent1_final = await agent1.generate_response(prompt3, session) print(f"{agent1_final}\n") print(f"✅ 第{i}场辩论结束\n") # 短暂延迟，避免API限制 await asyncio.sleep(1) print(f"\n🎉 八仙辩论全部结束！") print(f"📊 共进行了 {len(self.debate_pairs)} 场对角线辩论") print(f"🎭 参与仙人: {', '.join(self.agents.keys())}") async def main(): """主函数""" # 配置 api_url = "http://master.tailnet-68f9.ts.net:40012" # 尝试不同的API密钥格式 gemini_key = os.getenv('GEMINI_API_KEY', '') if not gemini_key: print("❌ 错误: 未找到GEMINI_API_KEY环境变量") print("请设置环境变量: export GEMINI_API_KEY=your_api_key") return # 使用提供的LiteLLM虚拟密钥 test_keys = [ "sk-0jdcGHZJpX2oUJmyEs7zVA" # LiteLLM虚拟密钥 ] print("🚀 启动八仙自定义API辩论系统...") # 辩论主题 topic = "工作量证明vs无限制爬虫：从李时珍采药到AI数据获取的激励机制变革" # 尝试不同的API密钥 for api_key in test_keys: if not api_key or api_key == "sk-": continue print(f"\n🔑 尝试API密钥: {api_key[:15]}...") debate_system = BaXianCustomDebateSystem(api_url, api_key) # 测试连接 if await debate_system.test_api_connection(): print(f"✅ 使用API密钥成功: {api_key[:15]}...") await debate_system.conduct_debate(topic) return else: print(f"❌ API密钥失败: {api_key[:15]}...") print("\n❌ 所有API密钥都失败了") print("\n🔍 可能的解决方案:") print(" 1. 检查LiteLLM服务器是否正确配置") print(" 2. 确认API密钥格式") print(" 3. 联系服务器管理员获取正确的虚拟密钥") print(" 4. 检查网络连接和防火墙设置") if __name__ == "__main__": asyncio.run(main())