本课程是关于
大模型智能体
的实战课程，包括原理、算法、应用场景、代码实战案例等，下表是本次课程的大纲。本课是第
26
节，讲解3个智能体参与实时辩论
。本课约
4200
字，阅读时长
20min
。下面音频是本课的一个辩论的音频录音，读者可以先听听效果。
以下是本次课程的正文：
前面的几个案例我们都是通过大模型驱动一个智能体来完成某项任务，体现的是单个智能体的能力。就如
第23课
所讲，多个智能体是能进行协作、竞争的，通过多个智能体能实现群体智能，因而能解决一些单个智能体解决不了或者解决不好的问题。
本课我们就通过辩论这种方式来尝试一个多智能体的小项目，我们先讲解辩论的整体架构、然后实现辩论过程需要的实时TTS（Text To Speech）和实时播放，最后实现智能体的辩论流程。
辩论中最重要的是参与方及辩论规则。本案例我们有3个参与方：正方、反方、裁判，每个参与方都是一个智能体——都由大模型驱动，规则是事先设定的，在辩论开始之前裁判宣读，然后正反双方在基于规则之下进行辩论，维护自己观点、反驳对方观点。
为了让辩论过程更加有趣味、更加真实，我们会采用云端TTS将智能体的的发言（大模型生成的辩论文本）通过TTS技术实时转为语音，然后用本地播放器播放，这样在辩论过程中，智能体就像真人一样进行辩论了。
我们可以设定辩论的轮数，当正反两方完成对应的轮数时，由裁判基于辩论规则和正反双方的表现来确定输赢并给出理由——当然也是利用语音的方式给出的。具体的辩论架构见下面图26-1。
本案例中的智能体背后的大模型是使用的DeepSeek的模型，规则、对方的发言等信息会当做上下文注入智能体的Prompt中，然后利用大模型进行思考给出流式的文本回复。具体过程会在26.3中进行详细说明。下面我们先讲解实时TTS及实时语音播放的实现。
为了实现实时语音辩论，我们采用了minimax的TTS开放平台接口（见参考文献1）。这个接口支持设定不同的角色发音（我们案例中正反方、裁判分别采用了不同的声音，方便更好区别，也更真实），也支持流式调用，配合智能体的流式文本（即辩论的观点）输出，就可以模拟真实的辩论场景。具体的代码如下（对应github上代码仓库中的
llm_agent_abc/agent_debate/minimax_tts.py
）。
import jsonimport subprocessimport timefrom typing import Iteratorimport requestsfrom configs.model_config import MINIMAX_API_KEY, MINIMAX_URL
file_format = 'mp3'
def build_tts_stream_headers() -> dict: headers = { 'accept': 'application/json, text/plain, */*', 'content-type': 'application/json', 'authorization': "Bearer " + MINIMAX_API_KEY, }return headers
def build_tts_stream_body(text: str, voice_id: str) -> str: body = json.dumps({ "model": "speech-01-turbo", "text": text, "stream": True, "voice_setting": { "voice_id": voice_id, "speed": 1, "vol": 1, "pitch": 0, "emotion": "neutral" }, "pronunciation_dict": { "tone": [ "处理/(chu3)(li3)", "危险/dangerous" ] }, "audio_setting": { "audio_sample_rate": 32000, "bitrate": 128000, "format": file_format, "channel": 1 } }) return body
mpv_command = ["mpv", "--no-cache", "--no-terminal", "--", "fd://0"] mpv_process = subprocess.Popen( mpv_command, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,)
def call_tts_stream(text: str, voice_id: str) -> Iterator[bytes]: tts_headers = build_tts_stream_headers() tts_body = build_tts_stream_body(text, voice_id)
response = requests.request("POST", MINIMAX_URL, stream=True, headers=tts_headers, data=tts_body) for chunk in response.raw: if chunk: if chunk[:5] == b'data:': data = json.loads(chunk[5:]) if "data" in data and "extra_info" not in data: if "audio" in data["data"]: audio = data["data"]['audio'] yield audio
def audio_play(audio_stream: Iterator[bytes]) -> bytes: audio = b"" for chunk in audio_stream: if chunk is not None and chunk != '\n': decoded_hex = bytes.fromhex(chunk) mpv_process.stdin.write(decoded_hex) mpv_process.stdin.flush() audio += decoded_hex
return audio
if __name__ == "__main__": text = "真正的危险不是计算机开始像人一样思考，而是人开始像计算机一样思考。计算机只是可以帮我们处理一些简单事务。" voice_id = "female-tianmei" audio_chunk_iterator = call_tts_stream(text, voice_id) audio = audio_play(audio_chunk_iterator) timestamp = int(time.time()) file_name = f'output_total_{timestamp}.{file_format}' with open(file_name, 'wb') as file: file.write(audio)
上面代码比较简单，具体的一些参数有说明，没有说明的，读者可以从参考文献1中进行深入了解。为了在辩论中播放音频，我们用了一个跨平台的音频播放器
mpv。读者可以自行从参考文献2中进行了解和进行本地安装。
我们将上面的核心代码
call_tts_stream（生成tts语音）, audio_play（播放生成的音频文件）封装成了函数，方便在辩论过程中调用。下面讲解核心的辩论过程。
具体的辩论过程也不复杂。我们定义了一个Debate的类，这个类实现了整个辩论过程的控制（见下面图26-2），包括裁判宣读主题、规则，然后正反方轮流辩论及最终的结果宣读。
其中start函数中实现裁判宣读规则，然后利用一个while循环来控制整个辩论过程。
take_turn函数实现某方的逻辑，包括调用大模型流式生成发言文本、并实时播放，当完成发言后切换辩论方。最后函数final_judgment是裁判的宣读结果阶段，实现的逻辑也是通过大模型来基于规则和双方辩论过程文本，生成最终的结果及解释说明。具体代码读者可以参考下面（对应github代码仓库中的llm_agent_abc/agent_debate/debate.py）。
辩论过程中还用到了2个辅助函数fetch_streamed_text、
play_from_buffer（见下面代码），下面我们简单进行解释。
fetch_streamed_text是采用流式输出从大模型获取反馈，将参与方（正反方、裁判）的Prompt输入大模型，流式获取文本反馈，并将反馈（反馈是一个个token，可能是单个汉字、也可能是词组，还可能是标点符号）放入一个文本队列（
text_buffer
）中，方便播放语音时
从队列获取文本，将生成文本和声音播放解耦，提升辩论过程的体验。
