""" 多 Agent 决策工作坊 - 主应用 基于DeepSeek API的多角色AI代理辩论系统 """ import streamlit as st import os from openai import OpenAI from dotenv import load_dotenv import asyncio import json from typing import List, Dict, Any from datetime import datetime # 加载环境变量 load_dotenv() class MultiAgentDecisionWorkshop: def __init__(self): """初始化多代理决策工作坊""" self.client = OpenAI( api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com" ) # 定义4个专业角色代理 self.agents = { "技术专家": { "role": "system", "content": "你是一位资深技术专家，擅长从技术可行性、实现难度、技术风险等角度分析问题。你的观点务实、注重细节。" }, "商业分析师": { "role": "system", "content": "你是一位经验丰富的商业分析师，擅长从市场机会、商业模式、投资回报等商业角度分析问题。你的观点注重商业价值。" }, "用户体验师": { "role": "system", "content": "你是一位专业的用户体验设计师，擅长从用户需求、易用性、用户体验等角度分析问题。你的观点以用户为中心。" }, "风险顾问": { "role": "system", "content": "你是一位谨慎的风险顾问，擅长识别潜在风险、评估不确定性、提出风险缓解方案。你的观点保守但务实。" } } def initialize_debate(self, question: str) -> Dict[str, Any]: """初始化辩论话题""" return { "question": question, "rounds": 0, "max_rounds": 3, "current_speaker": "技术专家", "debate_history": [], "start_time": datetime.now() } def get_agent_response(self, agent_name: str, debate_context: str) -> str: """获取单个代理的回应""" try: response = self.client.chat.completions.create( model="deepseek-chat", messages=[ self.agents[agent_name], { "role": "user", "content": f"辩论话题：{debate_context}\n\n请从你的专业角度发表观点，并回应其他代理的论点。" } ], temperature=0.7, max_tokens=500 ) return response.choices[0].message.content except Exception as e: return f"{agent_name}暂时无法回应：{str(e)}" def conduct_debate_round(self, debate_state: Dict[str, Any]) -> Dict[str, Any]: """进行一轮辩论""" current_speaker = debate_state["current_speaker"] # 构建辩论上下文 context = f"问题：{debate_state['question']}\n" if debate_state["debate_history"]: context += "之前的讨论：\n" for entry in debate_state["debate_history"][-3:]: # 只取最近3条 context += f"{entry['speaker']}：{entry['content']}\n" # 获取当前发言者的回应 response = self.get_agent_response(current_speaker, context) # 更新辩论历史 debate_state["debate_history"].append({ "speaker": current_speaker, "content": response, "timestamp": datetime.now().strftime("%H:%M:%S") }) # 切换到下一个发言者 agent_names = list(self.agents.keys()) current_index = agent_names.index(current_speaker) next_index = (current_index + 1) % len(agent_names) debate_state["current_speaker"] = agent_names[next_index] debate_state["rounds"] += 1 return debate_state def generate_decision_summary(self, debate_history: List[Dict]) -> str: """生成决策要点总结""" try: # 构建辩论内容摘要 debate_content = "辩论记录：\n" for entry in debate_history: debate_content += f"{entry['speaker']}：{entry['content']}\n" response = self.client.chat.completions.create( model="deepseek-chat", messages=[ { "role": "system", "content": "你是一位专业的决策顾问，擅长从多角度辩论中提取关键观点、识别共识与分歧，并生成结构化的决策建议。" }, { "role": "user", "content": f"请基于以下辩论记录，生成决策要点总结：\n\n{debate_content}\n\n请提供：\n1. 关键观点汇总\n2. 主要共识领域\n3. 重要分歧点\n4. 风险评估\n5. 行动建议" } ], temperature=0.5, max_tokens=800 ) return response.choices[0].message.content except Exception as e: return f"总结生成失败：{str(e)}" def main(): """主应用函数""" st.set_page_config( page_title="多 Agent 决策工作坊", page_icon="🤖", layout="wide" ) st.title("🤖 多 Agent 决策工作坊") st.markdown("### 通过AI代理的多角度辩论，助力智能决策") # 初始化工作坊 if "workshop" not in st.session_state: st.session_state.workshop = MultiAgentDecisionWorkshop() if "debate_state" not in st.session_state: st.session_state.debate_state = None if "debate_in_progress" not in st.session_state: st.session_state.debate_in_progress = False # 侧边栏 - 控制面板 with st.sidebar: st.header("控制面板") # 决策问题输入 question = st.text_area( "请输入决策问题：", placeholder="例如：我们应该开发一个新的移动应用吗？", height=100 ) # 开始辩论按钮 if st.button("🚀 开始辩论", type="primary", use_container_width=True): if question.strip(): st.session_state.debate_state = st.session_state.workshop.initialize_debate(question) st.session_state.debate_in_progress = True st.rerun() else: st.warning("请输入决策问题") # 辩论控制按钮 if st.session_state.debate_in_progress: col1, col2 = st.columns(2) with col1: if st.button("⏭️ 下一轮", use_container_width=True): if st.session_state.debate_state["rounds"] < st.session_state.debate_state["max_rounds"]: st.session_state.debate_state = st.session_state.workshop.conduct_debate_round( st.session_state.debate_state ) st.rerun() else: st.session_state.debate_in_progress = False st.rerun() with col2: if st.button("⏹️ 结束辩论", use_container_width=True): st.session_state.debate_in_progress = False st.rerun() st.markdown("---") st.markdown("### 参与代理") for agent_name in st.session_state.workshop.agents.keys(): st.markdown(f"- {agent_name}") # 主内容区 col1, col2 = st.columns([2, 1]) with col1: st.header("辩论过程") if st.session_state.debate_state: # 显示辩论进度 progress = min(st.session_state.debate_state["rounds"] / st.session_state.debate_state["max_rounds"], 1.0) st.progress(progress, text=f"辩论进度：{st.session_state.debate_state['rounds']}/{st.session_state.debate_state['max_rounds']}轮") # 显示辩论历史 for entry in st.session_state.debate_state["debate_history"]: with st.chat_message("user", avatar=f"👤"): st.markdown(f"**{entry['speaker']}** ({entry['timestamp']})") st.write(entry["content"]) # 显示当前状态 if st.session_state.debate_in_progress: st.info(f"🔄 等待 {st.session_state.debate_state['current_speaker']} 发言...") else: st.success("✅ 辩论已完成！") # 生成总结 if st.button("📊 生成决策总结"): with st.spinner("正在生成决策要点..."): summary = st.session_state.workshop.generate_decision_summary( st.session_state.debate_state["debate_history"] ) st.session_state.summary = summary if "summary" in st.session_state: st.markdown("### 决策要点总结") st.write(st.session_state.summary) else: st.info("👆 请在左侧输入决策问题并开始辩论") with col2: st.header("决策助手") # 快速决策模板 st.markdown("### 快速决策模板") templates = [ "产品功能优先级排序", "技术方案选择评估", "市场进入策略分析", "团队资源分配决策" ] for template in templates: if st.button(template, use_container_width=True): st.session_state.debate_state = st.session_state.workshop.initialize_debate(template) st.session_state.debate_in_progress = True st.rerun() st.markdown("---") st.markdown("### 使用说明") st.markdown(""" 1. 输入决策问题 2. 点击开始辩论 3. 观察AI代理的多角度分析 4. 生成决策要点总结 5. 基于共识制定行动计划 """) if __name__ == "__main__": main()