树莓派遇上ChatGPT，魔法热线就此诞生！

尽管这种电话在几十年前就已过时,但许多人都对旋转拨号电话记忆犹新。这些旧电话，其实可以被改造成一个 ChatGPT 热线。这个由 Pollux Labs 开发的项目，让你可以将一部复古的旋转拨号电话连接到树莓派上，拿起听筒、拨号，就能享受由 AI 驱动的对话，仿佛回到了传统的电话时代。

树莓派负责语音识别、文本生成和语音播放，ChatGPT 会记住通话中的每一句话。这意味着你可以体验到将老式拨号与尖端人工智能相结合的独特互动。现在，让我们来看看这是如何实现的。

将旋转电话改造成 ChatGPT 热线的理由

许多人喜欢使用旋转电话的复古感，尽管通话内容是现代的，但它的拨号声和重量能把你带回过去。ChatGPT 增添了有趣且由语音驱动的体验，与在键盘上打字完全不同。你还可以欣赏将电话的扬声器、麦克风和拨号盘连接到树莓派的工程挑战。除此之外，这是一个有趣的方式，可以重用旧技术。

通过电话听到 ChatGPT 的回应可以激发创造力。你可以在转移到语音助手之前，整合音乐、新闻更新或引入其他 AI 服务。实践是学习的关键，这个项目同时探索了硬件和软件。它展示了简单电子设备的灵活性。最重要的是，旋转拨号与 AI 对话会让任何尝试过它的人感到惊喜和愉悦。

电话改造成 ChatGPT 热线所需的必备物品

首先，你需要一部有足够的空间容纳树莓派和电线的旋转拨号电话。70 年代或 80 年代的电话型号通常内部空间较大，你可以整理电线而无需钻孔。你至少需要一台树莓派 4B，但树莓派 5 的性能会更好。

你还需要一个麦克风来捕捉音频，并将树莓派的音频输出连接到电话的扬声器。一个 USB 领夹麦克风或小型 USB 麦克风适配器应该可以完美地安装在机壳内部。

你可能会问，为什么要使用领夹麦克风而不是电话听筒中内置的麦克风。事实证明，尝试使用听筒的麦克风很困难，特别是因为旋转拨号电话中使用的麦克风是模拟的而不是数字的。

接下来，收集必要的电子工具，如烙铁、剪线钳和万用表。这些工具可以帮助你确认拨号的脉冲线、测试连接，并将树莓派的音频输出连接到电话的扬声器线。你还需要与树莓派 GPIO 引脚匹配的跳线或连接器，可能还需要一个小按钮来检测听筒是在线还是离线。

在软件方面，安装用于语音识别、文本转语音和 OpenAI API 的 Python 库。获取 OpenAI API 密钥，并在你的 Python 脚本中引用它，以生成 ChatGPT 回复。

完成改造并构建 ChatGPT 热线的步骤

将电话和树莓派改造成新用途涉及仔细的接线和软件配置。在此指南中，你将学习如何拆卸电话、识别拨号脉冲，并设置树莓派以实现语音转文本和文本转语音转换。仔细验证每根电线和引脚分配，因为一个不匹配可能会导致错误。

1. 取下电话盖，找到扬声器线、旋转拨号线，以及任何可以连接按钮以检测挂钩状态的地方。

2. 剥去 3.5 毫米音频电缆的外皮，将 2.8 毫米平板连接器焊接到电话听筒的地线和一个声道线上。然后，将其连接到听筒的连接插座。

3. 在电话内部放置一个 USB 麦克风（或适配器），确保你的树莓派可以清晰地接收声音。

4. 使用万用表确认哪些拨号线承载脉冲。将这些线连接到 GPIO 引脚和地线。然后，连接挂钩按钮，使软件能够感应到何时提起听筒。

5. 在你的树莓派 上安装必要的音频库，包括 PyAudio、PyGame 和 OpenAI 客户端。下载或创建音频文件（如拨号音）以供播放，并将你的 OpenAI 密钥存储在 .env 文件中。

6. 接下来，你需要一个 Python 脚本，用于从麦克风捕获音频，将其发送到 ChatGPT 进行处理，并通过电话扬声器播放 AI 的回应。你可以编写自己的脚本或使用 Pollux Labs 编写的脚本。只需确保根据自己的需求调整 GPIO 引脚编号、音频设置和特殊文本提示。

7. 手动运行脚本以确认其正常工作。一旦你听到拨号音且 ChatGPT 对你的声音做出回应，添加一个系统服务，以便在树莓派启动时自动启动电话。

如果你无法访问脚本，以下是供你参考的脚本。

#!/usr/bin/env python3"""ChatGPT for Rotary Phonehttps://en.polluxlabs.netMIT LicenseCopyright (c) 2025 Frederik KumbartzkiPermission is hereby granted, free of charge, to any person obtaining a copyof this software and associated documentation files (the "Software"), to dealin the Software without restriction, including without limitation the rightsto use, copy, modify, merge, publish, distribute, sublicense, and/or sellcopies of the Software, and to permit persons to whom the Software isfurnished to do so, subject to the following conditions:The above copyright notice and this permission notice shall be included in allcopies or substantial portions of the Software.THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS ORIMPLIED,

INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THEAUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHERLIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THESOFTWARE."""import osimport sysimport timeimport threadingfrom queue import Queuefrom pathlib import PathAudio and speech librariesos.environ['PYGAME_HIDE_SUPPORT_PROMPT'] ="hide"import pygameimport pyaudioimport numpy as npimport wavefrom openai import OpenAIOpenAI API Keyfrom dotenv import load_dotenvload_dotenv()OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")if not OPENAI_API_KEY:print("Error: OPENAI_API_KEY not found.")sys.exit(1)Hardware librariesfrom gpiozero import ButtonConstants and configurationsAUDIO_DIR ="/home/pi/Desktop/callGPT"AUDIO_FILES =

{"tone": f"{AUDIO_DIR}/a440.mp3","try_again": f"{AUDIO_DIR}/tryagain.mp3","error": f"{AUDIO_DIR}/error.mp3"}DIAL_PIN = 23# GPIO pin for rotary dialSWITCH_PIN = 17# GPIO pin for hook switchAudio parametersAUDIO_FORMAT = pyaudio.paInt16CHANNELS = 1SAMPLE_RATE = 16000CHUNK_SIZE = 1024SILENCE_THRESHOLD = 500MAX_SILENCE_CHUNKS = 20# About 1.3 seconds of silenceDEBOUNCE_TIME = 0.1# Time in seconds for debouncing button inputsclass AudioManager:"""Manages audio playback and recording."""def__init__(self):pygame.mixer.init(frequency=44100, buffer=2048)self.playing_audio = Falseself.audio_thread = NoneCreate temp directoryself.temp_dir

= Path(__file__).parent /"temp_audio"self.temp_dir.mkdir(exist_ok=True)Preload soundsself.sounds = {}for name, path in AUDIO_FILES.items():try:self.sounds[name] = pygame.mixer.Sound(path)except:print(f"Error loading {path}")def play_file(self, file_path, wait=True):try:sound = pygame.mixer.Sound(file_path)channel = sound.play()if wait and channel:while channel.get_busy():pygame.time.Clock().tick(30)except:pygame.mixer.music.load(file_path)pygame.mixer.music.play()if wait:while pygame.mixer.music.get_busy():pygame.time.Clock().tick(30)def start_continuous_tone(self):self.playing_audio = Trueif self.audio_thread and self.audio_thread.is_alive():self.playing_audio = Falseself.audio_thread.join(timeout=1.0)self.audio_thread = threading.Thread(target=self._play_continuous_tone)self.audio_thread.daemon = Trueself.audio_thread.start()def _play_continuous_tone(self):try:if"tone"in self.sounds:self.sounds["tone"].play(loops=-1)while self.playing_audio:time.sleep(0.1)self.sounds["tone"].stop()else:pygame.mixer.music.load(AUDIO_FILES["tone"])pygame.mixer.music.play(loops=-1)while self.playing_audio:time.sleep(0.1)pygame.mixer.music.stop()except Exception as e:print(f"Error during tone playback: {e}")def stop_continuous_tone(self):self.playing_audio = Falseif"tone"in self.sounds:self.sounds["tone"].stop()if pygame.mixer.get_init() and pygame.mixer.music.get_busy():pygame.mixer.music.stop()class SpeechRecognizer:"""Handles real-time speech recognition using OpenAI's Whisper API."""def__init__(self, openai_client):self.client = openai_clientself.audio = pyaudio.PyAudio()self.stream = Nonedef capture_and_transcribe(self):Setup audio stream if not already initializedif not self.stream:self.stream = self.audio.open(format=AUDIO_FORMAT,channels=CHANNELS,rate=SAMPLE_RATE,input=True,frames_per_buffer=CHUNK_SIZE,)Set up queue and threadingaudio_queue = Queue()stop_event = threading.Event()Start audio capture threadcapture_thread = threading.Thread(target=self._capture_audio,args=(audio_queue, stop_event))capture_thread.daemon = Truecapture_thread.start()Process the audioresult = self._process_audio(audio_queue, stop_event)Cleanupstop_event.set()capture_thread.join()return resultdef _capture_audio(self, queue, stop_event):while not stop_event.is_set():try:data = self.stream.read(CHUNK_SIZE, exception_on_overflow=False)queue.put(data)except KeyboardInterrupt:breakdef _process_audio(self, queue, stop_event):buffer = b""speaking = Falsesilence_counter = 0while not stop_event.is_set():if not queue.empty():chunk = queue.get()Check volumedata_np = np.frombuffer(chunk, dtype=np.int16)volume = np.abs(data_np).mean()Detect speakingif volume > SILENCE_THRESHOLD:speaking = Truesilence_counter = 0elif speaking:silence_counter += 1Add chunk to bufferbuffer += chunkProcess if we've detected end of speechif speaking and silence_counter > MAX_SILENCE_CHUNKS:print("Processing speech...")Save to temp filetemp_file = Path(__file__).parent /"temp_recording.wav"self._save_audio(buffer, temp_file)Transcribetry:return self._transcribe_audio(temp_file)except Exception as e:print(f"Error during transcription: {e}")buffer = b""speaking = Falsesilence_counter = 0return Nonedef _save_audio(self, buffer, file_path):with wave.open(str(file_path),"wb") as

wf:wf.setnchannels(CHANNELS)wf.setsampwidth(self.audio.get_sample_size(AUDIO_FORMAT))wf.setframerate(SAMPLE_RATE)wf.writeframes(buffer)def _transcribe_audio(self, file_path):with open(file_path,"rb") as audio_file:transcription = self.client.audio.transcriptions.create(model="whisper-1",file=audio_file,language="en")return transcription.textdef cleanup(self):if self.stream:self.stream.stop_stream()self.stream.close()self.stream = Noneif self.audio:self.audio.terminate()self.audio = Noneclass ResponseGenerator:"""Generates and speaks streaming responses from OpenAI's API."""def__init__(self, openai_client, temp_dir):self.client = openai_clientself.temp_dir = temp_dirself.answer =""def generate_streaming_response(self, user_input, conversation_history=None):self.answer =""collected_messages = []chunk_files = []Audio playback queue and control variablesaudio_queue = Queue()playing_event = threading.Event()stop_event = threading.Event()Start the audio playback threadplayback_thread = threading.Thread(target=self._audio_playback_worker,args=(audio_queue, playing_event, stop_event))playback_thread.daemon = Trueplayback_thread.start()Prepare messagesmessages = [{"role":"system","content":"You are a humorous conversation partner engaged in a natural phone call. Keep your answers concise and to the point."}]Use conversation history if available, but limit to last 4 pairsif conversation_history and len(conversation_history) > 0:if len(conversation_history) > 8:conversation_history = conversation_history[-8:]messages.extend(conversation_history)else:messages.append({"role":"user","content": user_input})Stream the responsestream = self.client.chat.completions.create(model="gpt-4o-mini",messages=messages,stream=True)Variables for sentence chunkingsentence_buffer =""chunk_counter = 0for chunk in stream:if chunk.choices and hasattr(chunk.choices[0], 'delta') and hasattr(chunk.choices[0].delta, 'content'):content = chunk.choices[0].delta.contentif content:collected_messages.append(content)sentence_buffer += contentProcess when we have a complete sentence or phraseif any(end in content for end in [".","!","?",":"]) or len(sentence_buffer) > 100:Generate speech for this chunkchunk_file_path = self.temp_dir / f"chunk_{chunk_counter}.mp3"try:Generate speechresponse = self.client.audio.speech.create(model="tts-1",voice="alloy",input=sentence_buffer,speed=1.0)response.stream_to_file(str(chunk_file_path))chunk_files.append(str(chunk_file_path))Add to playback queueaudio_queue.put(str(chunk_file_path))Signal playback thread if it's waitingplaying_event.set()except Exception as e:print(f"Error generating speech for chunk: {e}")Reset buffer and increment countersentence_buffer =""chunk_counter += 1Process any remaining textif sentence_buffer.strip():chunk_file_path = self.temp_dir / f"chunk_{chunk_counter}.mp3"try:response = self.client.audio.speech.create(model="tts-1",voice="alloy",input=sentence_buffer,speed=1.2)response.stream_to_file(str(chunk_file_path))chunk_files.append(str(chunk_file_path))audio_queue.put(str(chunk_file_path))playing_event.set()except Exception as e:print(f"Error generating final speech chunk: {e}")Signal end of generationaudio_queue.put(None)# Sentinel to signal end of queueWait for playback to completeplayback_thread.join()stop_event.set()# Ensure the thread stopsCombine all messagesself.answer ="".join(collected_messages)print(self.answer)Clean up temp filesself._cleanup_temp_files(chunk_files)return self.answerdef _audio_playback_worker(self, queue, playing_event, stop_event):while not stop_event.is_set():Wait for a signal that there's something to playif queue.empty():playing_event.wait(timeout=0.1)playing_event.clear()continueGet the next file to playfile_path = queue.get()None is our sentinel value to signal end of queueif file_path is None:breaktry:Play audio and wait for completionpygame.mixer.music.load(file_path)pygame.mixer.music.play()Wait for playback to complete before moving to next chunkwhile pygame.mixer.music.get_busy() and not stop_event.is_set():pygame.time.Clock().tick(30)Small pause between chunks for more natural flowtime.sleep(0.05)except Exception as e:print(f"Error playing audio chunk: {e}")def _cleanup_temp_files(self, file_list):Wait a moment to ensure files aren't in usetime.sleep(0.5)for file_path in file_list:try:if os.path.exists(file_path):os.remove(file_path)except Exception as e:print(f"Error removing temp file: {e}")class RotaryDialer:"""Handles rotary phone dialing and services."""def__init__(self, openai_client):self.client = openai_clientself.audio_manager = AudioManager()self.speech_recognizer = SpeechRecognizer(openai_client)self.response_generator = ResponseGenerator(openai_client, self.audio_manager.temp_dir)Set up GPIOself.dial_button = Button(DIAL_PIN, pull_up=True)self.switch = Button(SWITCH_PIN, pull_up=True)State variablesself.pulse_count = 0self.last_pulse_time = 0self.running = Truedef start(self):Set up callbacksself.dial_button.when_pressed = self._pulse_detectedself.switch.when_released = self._handle_switch_releasedself.switch.when_pressed = self._handle_switch_pressedStart in ready stateif not self.switch.is_pressed:Receiver is picked upself.audio_manager.start_continuous_tone()else:Receiver is on hookprint("Phone in idle state. Pick up the receiver to begin.")print("Rotary dial ready. Dial a number when the receiver is picked up.")try:self._main_loop()except KeyboardInterrupt:print("Terminating...")self._cleanup()def _main_loop(self):while self.running:self._check_number()time.sleep(0.1)def _pulse_detected(self):if not self.switch.is_pressed:current_time = time.time()if current_time - self.last_pulse_time >

DEBOUNCE_TIME:self.pulse_count += 1self.last_pulse_time = current_timedef _check_number(self):if not self.switch.is_pressed and self.pulse_count > 0:self.audio_manager.stop_continuous_tone()time.sleep(1.5)# Wait between digitsif self.pulse_count == 10:self.pulse_count = 0# "0" is sent as 10 pulsesprint("Dialed service number:", self.pulse_count)if self.pulse_count == 1:self._call_gpt_service()Return to dial tone after conversationif not self.switch.is_pressed:

# Only if the receiver wasn't hung upself._reset_state()self.pulse_count = 0def _call_gpt_service(self):Conversation history for contextconversation_history = []first_interaction = TrueFor faster transitionsspeech_recognizer = self.speech_recognizerresponse_generator = self.response_generatorPreparation for next recordingnext_recording_thread = Nonenext_recording_queue = Queue()Conversation loop - runs until the receiver is hung upwhile not self.switch.is_pressed:If there's a prepared next recording thread, use its resultif next_recording_thread:next_recording_thread.join()recognized_text = next_recording_queue.get()next_recording_thread = Noneelse:Only during first iteration or as fallbackprint("Listening..."+ (" (Speak now)"if first_interactionelse""))

first_interaction = FalseStart audio processingrecognized_text = speech_recognizer.capture_and_transcribe()if not recognized_text:print("Could not recognize your speech")self.audio_manager.play_file(AUDIO_FILES["try_again"])continueprint("Understood:", recognized_text)Update conversation historyconversation_history.append({"role":"user","content": recognized_text

})

Start the next recording thread PARALLEL to API responsenext_recording_thread = threading.Thread(target=self._background_capture,args=(speech_recognizer, next_recording_queue))next_recording_thread.daemon = Truenext_recording_thread.start()Generate the responseresponse = response_generator.generate_streaming_response(recognized_text, conversation_history)Add response to historyconversation_history.append({"role":"assistant","content": response})Check if the receiver was hung up in the meantimeif self.switch.is_pressed:breakIf we get here, the receiver was hung upif next_recording_thread and next_recording_thread.is_alive():next_recording_thread.join(timeout=0.5)def _background_capture(self, recognizer, result_queue):try:result = recognizer.capture_and_transcribe()result_queue.put(result)except Exception as e:print

(f"Error in background recording: {e}")result_queue.put(None)def _reset_state(self):self.pulse_count = 0self.audio_manager.stop_continuous_tone()self.audio_manager.start_continuous_tone

()print("Rotary dial ready. Dial a number.")def _handle_switch_released(self):print("Receiver picked up - System restarting")self._restart_script()def _handle_switch_pressed(self):print("Receiver hung up - System

terminating")self._cleanup()self.running = FalseComplete termination after short delaythreading.Timer(1.0, self._restart_script).start()returndef _restart_script(self):print("Script restarting...")self.audio_manager.stop_continuous_tone()os.execv(sys.executable, ['python'] + sys.argv)def _cleanup(self):Terminate Audio Managerself.audio_manager.stop_continuous_tone()Terminate Speech Recognizer if it existsif hasattr(self, 'speech_recognizer') and self.speech_recognizer:self.speech_recognizer.cleanup()print("Resources have been released.")def main():Initialize OpenAI clientclient = OpenAI(api_key=OPENAI_API_KEY)Create and start the rotary dialerdialer = RotaryDialer(client)dialer.start

()print("Program terminated.")ifname=="__main__":main()

回顾你所做的任何连接或配置的优化。每种电话型号都有细微差别，因此你可能需要进行一些实验。在电话外壳内给树莓派通电，以简化故障排除，直到你确信一切正常。一旦你通过听筒听到 ChatGPT 的回应，你的旋转电话热线就几乎完成了。

享受复古体验，同时访问现代 AI

用 ChatGPT 为旧式旋转电话注入活力，将怀旧与创新融为一体。拨打电话给 AI，展示了过去与现在技术之间的神奇互动。随着时间的推移，你可以通过不同的声音、语言或提示来个性化你的脚本，以改变 ChatGPT 的回应。

你甚至可以整合更多的 AI 服务，用于阅读新闻、播放播客或安排日程。这个项目可以让你接触电子设备和 Python 编程，因为你正在弥合模拟电话和数字 AI 之间的鸿沟。完成此项目后，你将拥有一部功能齐全的旋转电话，它充当 ChatGPT 的热线。享受你的复古未来主义对话，并探索为你的电话 AI 伙伴的新想法。

参考文章:

https://www.xda-developers.com/take-chatgpt-retro-raspberry-pi-powered-rotary-phone-hotline/

如果觉得文章不错记得点赞，收藏，关注，转发~