import streamlit as st import asyncio import sys from pathlib import Path from typing import Dict, Any, List # Ensure the main project directory is in the Python path project_root = Path(__file__).parent.parent.parent sys.path.insert(0, str(project_root)) try: from google.adk import Agent, Runner from google.adk.sessions import InMemorySessionService, Session from google.genai import types ADK_AVAILABLE = True except ImportError: ADK_AVAILABLE = False # 创建占位符类 class Agent: pass class Runner: pass class InMemorySessionService: pass class Session: pass class types: class Content: pass class Part: pass async def _get_llm_reply(runner: Runner, session: Session, prompt: str) -> str: """Helper function to call a Runner and get a text reply.""" content = types.Content(role='user', parts=[types.Part(text=prompt)]) response = runner.run_async( user_id=session.user_id, session_id=session.id, new_message=content ) reply = "" async for event in response: if hasattr(event, 'content') and event.content and hasattr(event.content, 'parts'): for part in event.content.parts: if hasattr(part, 'text') and part.text: reply += str(part.text) elif hasattr(event, 'text') and event.text: reply += str(event.text) return reply.strip() async def run_adk_debate_streamlit(topic: str, participants: List[str], rounds: int): """ Runs the ADK turn-based debate and yields each statement for Streamlit display. """ try: yield "🚀 **启动ADK八仙轮流辩论 (太上老君主持)...**" all_immortals = ["铁拐李", "吕洞宾", "何仙姑", "张果老", "蓝采和", "汉钟离", "韩湘子", "曹国舅"] if not participants: participants = all_immortals character_configs = { "太上老君": {"name": "太上老君", "model": "gemini-1.5-pro", "instruction": "你是太上老君，天道化身，辩论的主持人。你的言辞沉稳、公正、充满智慧。你的任务是：1. 对辩论主题进行开场介绍。2. 在每轮开始时进行引导。3. 在辩论结束后，对所有观点进行全面、客观的总结。保持中立，不偏袒任何一方。"}, "铁拐李": {"name": "铁拐李", "model": "gemini-1.5-flash", "instruction": "你是铁拐李，八仙中的逆向思维专家。你善于从批判和质疑的角度看问题，发言风格直接、犀利，但富有智慧。"}, "吕洞宾": {"name": "吕洞宾", "model": "gemini-1.5-flash", "instruction": "你是吕洞宾，八仙中的理性分析者。你善于平衡各方观点，用理性和逻辑来分析问题，发言风格温和而深刻。"}, "何仙姑": {"name": "何仙姑", "model": "gemini-1.5-flash", "instruction": "你是何仙姑，八仙中的风险控制专家。你总是从风险管理的角度思考问题，善于发现潜在危险，发言风格谨慎、细致。"}, "张果老": {"name": "张果老", "model": "gemini-1.5-flash", "instruction": "你是张果老，八仙中的历史智慧者。你善于从历史数据中寻找规律和智慧，提供长期视角，发言风格沉稳、博学。"}, "蓝采和": {"name": "蓝采和", "model": "gemini-1.5-flash", "instruction": "你是蓝采和，八仙中的创新思维者。你善于从新兴视角和非传统方法来看待问题，发言风格活泼、新颖。"}, "汉钟离": {"name": "汉钟离", "model": "gemini-1.5-flash", "instruction": "你是汉钟离，八仙中的平衡协调者。你善于综合各方观点，寻求和谐统一的解决方案，发言风格平和、包容。"}, "韩湘子": {"name": "韩湘子", "model": "gemini-1.5-flash", "instruction": "你是韩湘子，八仙中的艺术感知者。你善于从美学和感性的角度分析问题，发言风格优雅、感性。"}, "曹国舅": {"name": "曹国舅", "model": "gemini-1.5-flash", "instruction": "你是曹国舅，八仙中的实务执行者。你关注实际操作和具体细节，发言风格务实、严谨。"} } session_service = InMemorySessionService() session = await session_service.create_session(state={}, app_name="稷下学宫八仙论道系统-Streamlit", user_id="st_user") runners: Dict[str, Runner] = {} for name, config in character_configs.items(): if name == "太上老君" or name in participants: agent = Agent(name=config["name"], model=config["model"], instruction=config["instruction"]) runners[name] = Runner(app_name="稷下学宫八仙论道系统-Streamlit", agent=agent, session_service=session_service) host_runner = runners.get("太上老君") if not host_runner: yield "❌ **主持人太上老君初始化失败。**" return yield f"🎯 **参与仙人**: {', '.join(participants)}" debate_history = [] # Opening statement opening_prompt = f"请为本次关于“{topic}”的辩论，发表一段公正、深刻的开场白，并宣布辩论开始。" opening_statement = await _get_llm_reply(host_runner, session, opening_prompt) yield f"👑 **太上老君**: {opening_statement}" # Debate rounds for round_num in range(rounds): round_intro_prompt = f"请为第 {round_num + 1} 轮辩论说一段引导语。" round_intro = await _get_llm_reply(host_runner, session, round_intro_prompt) yield f"👑 **太上老君**: {round_intro}" for name in participants: if name not in runners: continue history_context = f"\n最近的论道内容：\n" + "\n".join([f"- {h}" for h in debate_history[-5:]]) if debate_history else "" prompt = f"论道主题: {topic}{history_context}\n\n请从你的角色特点出发，简洁地发表观点。" reply = await _get_llm_reply(runners[name], session, prompt) yield f"🗣️ **{name}**: {reply}" debate_history.append(f"{name}: {reply}") await asyncio.sleep(1) # Summary summary_prompt = f"辩论已结束。以下是完整的辩论记录：\n\n{' '.join(debate_history)}\n\n请对本次辩论进行全面、公正、深刻的总结。" summary = await _get_llm_reply(host_runner, session, summary_prompt) yield f"👑 **太上老君**: {summary}" for runner in runners.values(): await runner.close() yield "🎉 **ADK八仙轮流辩论完成!**" except Exception as e: yield f"❌ **运行ADK八仙轮流辩论失败**: {e}" import traceback st.error(traceback.format_exc()) def render_adk_debate_tab(): """Renders the Streamlit UI for the ADK Debate tab.""" # 检查 ADK 是否可用 if not ADK_AVAILABLE: st.error("🚫 Google ADK 模块未安装或不可用") st.info("📦 正在安装 Google ADK，请稍候...") st.info("💡 安装完成后请刷新页面") with st.expander("📋 安装说明"): st.code(""" # 安装 Google ADK pip install google-adk>=1.12.0 # 或从 GitHub 安装开发版 pip install git+https://github.com/google/adk-python.git@main """) return st.markdown("### 🏛️ 八仙论道 (ADK版 - 太上老君主持)") topic = st.text_input( "辩论主题", value="AI是否应该拥有创造力？", key="adk_topic_input" ) all_immortals = ["铁拐李", "吕洞宾", "何仙姑", "张果老", "蓝采和", "汉钟离", "韩湘子", "曹国舅"] col1, col2 = st.columns(2) with col1: rounds = st.number_input("辩论轮数", min_value=1, max_value=5, value=1, key="adk_rounds_input") with col2: participants = st.multiselect( "选择参与的仙人 (默认全选)", options=all_immortals, default=all_immortals, key="adk_participants_select" ) if st.button("🚀 开始论道", key="start_adk_debate_button", type="primary"): if not topic: st.error("请输入辩论主题。") return if not participants: st.error("请至少选择一位参与的仙人。") return st.markdown("---") st.markdown("#### 📜 论道实录") # Placeholder for real-time output output_container = st.empty() full_log = "" # Run the async debate function try: # Get a new event loop for the thread loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) # Create a coroutine object coro = run_adk_debate_streamlit(topic, participants, rounds) # Run the coroutine until completion for message in loop.run_until_complete(async_generator_to_list(coro)): full_log += message + "\n\n" output_container.markdown(full_log) except Exception as e: st.error(f"启动辩论时发生错误: {e}") async def async_generator_to_list(async_gen): """Helper to consume an async generator and return a list of its items.""" return [item async for item in async_gen]