250 lines
10 KiB
Python
250 lines
10 KiB
Python
#!/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()) |