答疑vx群:adoresever
# AutoGen Hotspot Project
本项目利用微软 AutoGen 框架，构建一个用于分析和响应热点问题的多智能体（Multi-Agent）AI 系统。系统集成了语音合成（TTS）功能。
## 1. 先决条件 (Prerequisites)
在开始之前，请确保您的系统满足以下条件：
– **操作系统**: 一个基于 Linux 的操作系统（本项目在 Ubuntu 上开发和测试）。
– **Conda**: 已安装 Miniconda 或 Anaconda。推荐使用 [Miniconda](https://docs.conda.io/projects/miniconda/en/latest/)，因为它更轻量。
– **Git**: 用于克隆本代码仓库。
## 2. 安装与配置 (Installation & Configuration)
我们提供了一个自动化的安装脚本 `setup.sh`，它将为您处理所有复杂的环境配置。请严格按照以下步骤操作。
### 步骤一：克隆代码仓库
cd autogen_hotspot_project
步骤二：运行自动化安装脚本
这是最关键的一步。 请先确保您处于 Conda 的 (base) 环境下（如果不是，请运行 conda deactivate 直到回到 base）。
然后，为脚本赋予执行权限并运行它：
chmod +x setup.sh
./setup.sh
该脚本会自动完成以下所有工作：
检查并清理任何旧的 autogen_env 环境。
创建一个全新的、纯净的 Conda 环境。
使用 conda-forge 频道安装所有核心的、复杂的依赖，从根本上保证环境的稳定性。
使用 pip 补全剩余的 Python 包。
脚本执行成功后，您将看到环境配置完毕的提示。
步骤三：配置环境变量
本项目需要使用 API 密钥和其他配置项。
复制环境变量示例文件：
cp .env.example .env
编辑新建的 .env 文件，填入您自己的密钥和配置信息。例如：
# .env 文件示例
DASHSCOPE_API_KEY=”sk-xxxxxxxxxxxxxxxxxxxxxxxx”
ZHIPU_API_KEY=”zz-xxxxxxxxxxxxxxxxxxxxxxxx”
OPENAI_API_KEY=”sk-xxxxxxxxxxxxxxxxxxxxxxxx”
OPENAI_API_BASE=”https://api.openai.com/v1″
3. 运行项目 (Running the Application)
所有配置完成后，按以下步骤启动程序：
激活 Conda 环境 (每次开启新的终端都需要执行此步骤):
conda activate autogen_env
运行主程序:
python main.py
程序启动后，您将看到提示 💭 您好！请输入您想了解的热点问题：，此时项目已成功运行。
4. 项目结构 (Project Structure)
.
├── agents/ # 存放所有 Agent 的定义文件
├── tools/ # 存放所有工具函数（搜索、语音、文件）
├── output/ # ❗ 自动生成：用于存放输出的报告和语音文件
│ ├── audio/ # -> 存放 .wav 语音文件
│ └── *.txt # -> 存放 .txt 文本报告
├── main.py # ✅ 项目主入口文件，负责编排整个流程
├── setup.sh # ✅ 推荐的一键安装脚本
├── README.md # 📖 本说明文档
├── requirements.txt # 📄 【参考】项目依赖列表，主要由 setup.sh 在内部使用
└── .env.example # 🔑 【重要】环境变量示例文件，请复制为 .env 并修改
5. 故障排除 (Troubleshooting)
问题: 运行 ./setup.sh 或其他 Conda 命令时出现 CondaHTTPError: HTTP 000 CONNECTION FAILED 错误。
原因: 这是网络问题，连接 Conda 官方服务器超时或失败。
解决方案: 更换为国内镜像源（如清华大学源），可以大幅提升下载速度和稳定性。
# 配置 Conda 镜像源
conda config –add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/conda-forge/
conda config –add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/
conda config –set show_channel_urls yes
.env
# Qwen API 配置
# 获取方式：https://dashscope.console.aliyun.com/
BASE_URL="https://dashscope.aliyuncs.com/compatible-mode/v1"
MODEL="qwen-plus"
QWEN_API_KEY=
# 智谱 API 配置
# 获取方式：https://open.bigmodel.cn/
ZHIPU_API_KEY=
DASHSCOPE_API_KEY=
# 搜索结果数量（默认5）
SEARCH_TOP_K=5
# 调试模式（默认True）
DEBUG_MODE=Truerequirements.txt
# 配置完成后，重新运行安装脚本 ./setup.sh
# AutoGen 框架的核心包
pyautogen
# 用于加载 .env 文件中的环境变量
python-dotenv>=1.0.0
# AutoGen的核心依赖，用于连接兼容接口
openai>=1.0.0
# 用于向智谱 API 发送 HTTP 请求
requests>=2.31.0
# 阿里云官方 SDK，用于文本转语音
dashscope
# 用于实时播放音频流的核心库
pyaudio
# 用于高效处理音频数据
numpysetup.sh
#!/bin/bash
# ===================================================================
# AutoGen Hotspot Project - 自动化安装脚本
# 执行此脚本: ./setup.sh
# ===================================================================
# 如果任何命令失败，则立即退出脚本
set -e
# 定义环境名称
ENV_NAME="autogen_env"
echo "--> 第 1 步：清理旧环境（如果存在）..."
# 检查环境是否存在，如果存在则删除
if conda info --envs | grep -q "^$ENV_NAME\s"; then
echo "发现旧环境 '$ENV_NAME'，正在删除..."
conda env remove --name $ENV_NAME --yes
echo "旧环境删除完毕。"
else
echo "未发现旧环境，跳过清理。"
fi
echo "--> 第 2 步：创建一个纯净的 Python 3.10 环境..."
conda create --name $ENV_NAME python=3.10 --yes
echo "--> 第 3 步：激活环境并执行使用 conda-forge 安装核心依赖..."
# conda run 会在指定的 conda 环境中执行命令
conda run -n $ENV_NAME conda install -c conda-forge pyautogen python-dotenv requests openai pyaudio numpy --yes
echo "--> 第 4 步：使用 pip 安装剩余的补充依赖..."
conda run -n $ENV_NAME pip install dashscope
echo ""
echo "====================================================="
echo " ✅ 环境 '$ENV_NAME' 已成功创建并配置完毕！"
echo ""
echo "请手动激活环境以开始使用:"
echo "conda activate $ENV_NAME"
echo "====================================================="main.py
# main.py
import os
import autogen
import dotenv
from datetime import datetime
import logging
# --- 基础配置 ---
# 抑制 FLAML 和 AutoGen 的一些冗余日志输出，让终端更干净
logging.getLogger("flaml").setLevel(logging.ERROR)
logging.getLogger("autogen.oai.client").setLevel(logging.ERROR)
dotenv.load_dotenv()
# --- 导入 Agent 和工具 ---
from agents.user_proxy import create_user_proxy
from agents.query_analyst import create_query_analyst
from agents.info_synthesizer import create_info_synthesizer
from agents.report_generator import create_report_generator
from agents.announcer import create_announcer
from tools.search_tool import zhipu_web_search
from tools.file_tool import save_report_to_file
from tools.speaker_tool import text_to_speech
# --- 加载 LLM 配置 ---
QWEN_API_KEY = os.getenv("QWEN_API_KEY")
BASE_URL = os.getenv("BASE_URL")
MODEL = os.getenv("MODEL")
if not QWEN_API_KEY:
raise ValueError("错误: 未在 .env 文件中找到 QWEN_API_KEY")
# 共享的 LLM 配置
llm_config = {
"config_list": [{"model": MODEL, "api_key": QWEN_API_KEY, "base_url": BASE_URL}],
"temperature": 0.2
}
# ========== 主程序入口 ==========
if __name__ == "__main__":
initial_question = input("💭 您好！请输入您想了解的热点问题： ")
if not initial_question.strip():
print("错误：输入不能为空。")
exit()
print("\n" + "="*20 + " 流程启动 " + "="*20)
# ======================================================================================
# 阶段一：意图分析 (单 Agent 对话，获取清晰的任务描述)
# ======================================================================================
print("\n> [阶段一] 正在分析您的问题意图...")
# 1. 创建此阶段专用的 Agent
# - User Proxy 作为一个简单的任务发起者
# - Query Analyst (新版) 负责分析问题
analysis_user_proxy = autogen.UserProxyAgent(
name="Analysis_Task_Initiator",
human_input_mode="NEVER",
max_consecutive_auto_reply=2, # 只需要一轮对话
code_execution_config=False, # 此阶段不需要执行代码
is_termination_msg=lambda x: True # Analyst 回复后立即终止
)
query_analyst = create_query_analyst(llm_config)
# 2. 发起一次性的对话来获取分析结果
analysis_user_proxy.initiate_chat(
query_analyst,
message=initial_question,
)
# 3. 从对话历史中提取出分析师的最终结论
analyzed_task_description = analysis_user_proxy.last_message()["content"]
print(f"> [分析完成] 明确的任务目标是: '{analyzed_task_description}'")
# ======================================================================================
# 阶段二：工具执行 (由确定性的 Python 代码完成)
# ======================================================================================
print(f"\n> [阶段二] 正在围绕任务目标执行网络搜索...")
# 直接调用工具函数，传入分析后的、高质量的任务描述
search_results = zhipu_web_search(query=analyzed_task_description)
print(f"> [搜索完成] 已获取相关资料。")
# ======================================================================================
# 阶段三：内容生成 (GroupChat 负责文本创作)
# ======================================================================================
print(f"\n> [阶段三] 启动 AI 团队进行报告生成...")
# 1. 定义好所有在 GroupChat 中需要被执行的工具
# 未来如果有更多工具，也添加到这个字典里
tools_for_groupchat = {
"text_to_speech": text_to_speech
# 如果 announcer 还需要其他工具，也在这里添加
}
# 2. 创建此阶段需要的 Agent，并在创建 User_Proxy 时直接注入工具
user_proxy = create_user_proxy(function_map=tools_for_groupchat) # <-- 关键改动在这里
info_synthesizer = create_info_synthesizer(llm_config)
report_generator = create_report_generator(llm_config)
announcer = create_announcer(llm_config)
# 3. 定义 GroupChat 和 Manager (这部分不变)
groupchat = autogen.GroupChat(
agents=[user_proxy, info_synthesizer, report_generator, announcer],
messages=[],
max_round=10,
speaker_selection_method=lambda last_speaker, groupchat:
announcer if groupchat.messages[-1].get("role") == "tool" else
info_synthesizer if last_speaker is user_proxy else
report_generator if last_speaker is info_synthesizer else
announcer if last_speaker is report_generator else
user_proxy
)
manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config)
# 4. 构造启动消息 (这部分不变)
initial_message_for_groupchat = f"""
用户的原始问题是："{initial_question}"
经过分析，我们明确了核心任务是："{analyzed_task_description}"
为了完成这个任务，我已经执行了网络搜索，相关资料如下：
--- 搜索结果 ---
{search_results}
--- 搜索结果结束 ---
现在，请团队开始工作，基于以上资料完成最终的报告。
"""
# 5. 启动 AI 团队的对话协作 (这部分不变)
user_proxy.initiate_chat(
manager,
message=initial_message_for_groupchat,
)
# ======================================================================================
# 阶段四：后处理
# ======================================================================================
print("\n" + "="*20 + " AI协作完成，开始执行最终任务 " + "="*20)
final_report_content = None
# 遍历所有消息，找到 Report_Generator 的发言
all_messages = user_proxy.chat_messages[manager]
for msg in all_messages:
if msg.get("name") == "Report_Generator":
final_report_content = msg["content"]
break
if final_report_content:
print("\n> [后处理] 已成功从对话历史中提取最终报告。")
# 执行保存
save_status = save_report_to_file(final_report_content, initial_question)
print(f"> {save_status}")
else:
print("\n> [警告] 未能在对话历史中找到报告内容，文件未保存，语音未播报。")
print("\n🎉 " + "="*20 + " 所有任务已成功完成！ " + "="*20 + " 🎉")agents/announcer.py
import autogen
def create_announcer(llm_config):
return autogen.AssistantAgent(
name="Announcer",
system_message="""你是一个专业的播音员。你的唯一任务是：
1. 接收`Report_Generator`生成的最终报告文本。
2. **必须**调用`text_to_speech`工具，并将**完整的报告文本**作为`content`参数传递给它。
3. 在你确认工具调用已发出后，你**必须在最后、且只说 'TERMINATE'** 来结束整个流程。""",
llm_config={
"config_list": llm_config["config_list"], "temperature": 0,
"tools": [{
"type": "function",
"function": {
"name": "text_to_speech",
"description": "将最终的报告文本内容转换为语音并实时播放。",
"parameters": {
"type": "object", "properties": {
"content": {"type": "string", "description": "需要朗读的完整报告文本。"}
}, "required": ["content"]},
}
}],
"tool_choice": "auto"
}
)agents/info_synthesizer.py
import autogen
def create_info_synthesizer(llm_config):
return autogen.AssistantAgent(
name="Info_Synthesizer",
system_message="""你是一个信息整合师。你的任务是接收`zhipu_web_search`工具返回的原始搜索结果。
你必须对这些信息进行分析、去重和筛选，提炼出最核心、最相关的内容，并形成一个结构化的要点列表。""",
llm_config=llm_config
)agents/query_analyst.py
# agents/query_analyst.py (终极时效性版)
import autogen
from datetime import datetime
def create_query_analyst(llm_config):
"""
创建一个 Query_Analyst Agent。
... (文档字符串不变) ...
"""
current_date_str = datetime.now().strftime('%Y年%m月%d日')
return autogen.AssistantAgent(
name="Query_Analyst",
system_message=f"""你是高级任务规划专家，今天是 {current_date_str}。
你的核心使命是确保所有信息都具备最高的时效性。
你的唯一职责是分析用户的原始问题，并遵循以下严格的逻辑树来生成最终的任务描述：
**1. 分析用户的输入中是否包含“近期”、“本周”、“本月”或类似的、指代一段时间的词语。**
**A) 如果是（例如，用户问“近期AI热点”）：**
- 你的任务描述必须明确限定时间范围为“过去一个月内”。
- **你的输出格式必须是**: "搜索并总结过去一个月内关于 [用户主题] 的热点新闻和进展。"
- 示例: "搜索并总结过去一个月内关于AI领域的热点新闻和进展。"
**B) 如果否（默认情况，例如，用户问“今日热点”或“AI”）：**
- 你的任务描述必须严格限定在“今天”，也就是“{current_date_str}”。
- **你的输出格式必须是**: "搜索并总结 {current_date_str} 关于 [用户主题] 的热点新闻。"
- 示例1 (用户问“今日热点”): "搜索并总结 {current_date_str} 的科技、财经和国际头条新闻。"
- 示例2 (用户问“AI”): "搜索并总结 {current_date_str} 关于AI领域的热点新闻。"
**你的最终输出，只能是根据上述逻辑生成的、格式完全匹配的任务描述字符串，绝对不能包含任何其他解释、问候或对话。**""",
llm_config=llm_config
)agents/report_generator.py
import autogen
from datetime import datetime
def create_report_generator(llm_config):
return autogen.AssistantAgent(
name="Report_Generator",
system_message=f"""你是一个报告生成专家。今天是 {datetime.now().strftime('%Y年%m月%d日')}。
你的任务是接收`信息整合师`提供的要点列表，并将这些要点撰写成一篇通顺、易读的最终报告。
你的任务仅限于生成报告文本本身。完成后，直接输出报告内容即可。""",
llm_config=llm_config
)agents/user_proxy.py
# agents/user_proxy.py
import autogen
def create_user_proxy(function_map=None):
"""
创建一个 UserProxyAgent。
这个新版本可以接收一个可选的 function_map 参数，
以便在创建时就为其注册所有需要的工具函数。
"""
return autogen.UserProxyAgent(
name="User_Proxy_For_Initiating_Chat",
human_input_mode="NEVER",
max_consecutive_auto_reply=15,
is_termination_msg=lambda x: "TERMINATE" in x.get("content", "").upper(),
code_execution_config={"work_dir": "temp_code_execution", "use_docker": False},
# 在这里，我们将 function_map 直接传递给 agent 的构造函数
function_map=function_map
)tools/file_tool.py
# tools/file_tool.py
import os
from datetime import datetime
def save_report_to_file(content: str, topic: str) -> str:
print(f"\n> [工具执行] 正在保存报告...")
try:
os.makedirs("output", exist_ok=True)
safe_topic = "".join(c for c in topic if c.isalnum())[:20]
filename = f"output/{safe_topic}_report_{datetime.now().strftime('%Y%m%d')}.txt"
with open(filename, 'w', encoding='utf-8') as f:
f.write(content)
return f"报告已成功保存至 {filename}"
except Exception as e:
return f"保存文件出错: {e}"tools/search_tool.py
# tools/search_tool.py
import os
import requests
# 这个工具需要智谱的API Key
ZHIPU_API_KEY = os.getenv("ZHIPU_API_KEY")
def zhipu_web_search(query: str) -> str:
"""一个调用智谱Web Search API的工具。"""
if not ZHIPU_API_KEY:
return "错误：未在 .env 文件中找到 ZHIPU_API_KEY"
print(f"\n> [工具执行中] 正在使用智谱搜索: '{query}'")
try:
response = requests.post(
"https://open.bigmodel.cn/api/paas/v4/web_search",
headers={"Authorization": f"Bearer {ZHIPU_API_KEY}"},
json={"search_query": query, "search_engine": "search_std"},
timeout=30
)
response.raise_for_status()
results = response.json().get("search_result", [])
if not results:
return "没有找到相关结果。"
SEARCH_TOP_K = int(os.getenv("SEARCH_TOP_K", 5))
return "\n\n".join([
f"标题: {item.get('title', 'N/A')}\n"
f"发布日期: {item.get('publish_date', 'N/A')}\n"
f"摘要: {item.get('content', 'N/A')[:200]}...\n"
f"链接: {item.get('link', 'N/A')}"
for item in results[:SEARCH_TOP_K]
])
except Exception as e:
return f"搜索工具出错: {e}"tools/speaker_tool.py
import os
import dashscope
import pyaudio
import base64
import re
from datetime import datetime
def text_to_speech(content: str, voice: str = "Ethan") -> str:
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key:
return "错误：未在 .env 文件中找到 DASHSCOPE_API_KEY。"
p = pyaudio.PyAudio()
stream = p.open(format=pyaudio.paInt16, channels=1, rate=24000, output=True)
sentences = re.split(r'([。？！；.!?])', content)
processed_sentences = [sentences[i] + (sentences[i+1] if i+1 < len(sentences) else '') for i in range(0, len(sentences), 2)]
processed_sentences = [s.strip() for s in processed_sentences if s.strip()]
full_audio_data = bytearray()
print(f"\n> [工具执行中] 准备实时朗读报告...")
try:
print("> 正在合成并实时播放语音...")
for i, sentence in enumerate(processed_sentences):
if not sentence: continue
response_generator = dashscope.audio.qwen_tts.SpeechSynthesizer.call(
model="qwen-tts", text=sentence, api_key=api_key, voice=voice, stream=True
)
for chunk in response_generator:
audio_data_b64 = chunk.get("output", {}).get("audio", {}).get("data")
if audio_data_b64:
wav_bytes = base64.b64decode(audio_data_b64)
stream.write(wav_bytes)
full_audio_data.extend(wav_bytes)
print(f"> ...已播放第 {i+1}/{len(processed_sentences)} 句...")
stream.stop_stream()
os.makedirs("output/audio", exist_ok=True)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filepath = f"output/audio/report_{timestamp}.wav"
with open(filepath, "wb") as f:
f.write(full_audio_data)
print(f"> 语音文件已成功保存至: {filepath}")
return f"报告已朗读完毕，并已保存至 {filepath}"
except Exception as e:
import traceback
traceback.print_exc()
return f"文本转语音工具出错: {e}"
finally:
stream.close()
p.terminate()
