實戰範例 002：自動化 Reddit 技術討論區輿情監控與 AI 回覆助理

簡介與場景

在現代軟體開發與社群運營中，自動化地監控社群平台（如 Reddit）的討論趨勢，並適時提供具有技術深度的回覆，對於開發者關係 (Developer Relations) 至關重要。 本範例將探討如何建構一個穩健的 Reddit 監控代理 (Monitor Agent)。我們將使用輪詢 (Polling) 或串流 (Streaming) 的方式獲取最新貼文，並結合大型語言模型 (Large Language Model, LLM) 自動生成回覆。 在極簡主義開發 (Ponytail) 的思維下，我們可能會忽略網路不穩定性與 API 速率限制 (Rate Limit)；然而，透過網路讀寫容錯 (Agent Reach) 的原則，我們將實作具備指數退避 (Exponential Backoff) 與請求重試機制 (Retry Mechanism) 的健壯系統。

原始代碼：過度簡單與脆弱的連線

以下是典型的初學者代碼，完全缺乏錯誤處理與狀態管理。一旦遇到 Reddit API 的 HTTP 429 Too Many Requests 錯誤，進程就會直接崩潰。

import praw
import time
from llm_client import generate_reply

# 脆弱的 Reddit 實例化，無超時或重試設定
reddit = praw.Reddit(
    client_id="YOUR_CLIENT_ID",
    client_secret="YOUR_CLIENT_SECRET",
    user_agent="my_bot/1.0"
)

def monitor_and_reply():
    subreddit = reddit.subreddit("programming")
    
    # 這裡的 stream 很容易因為網路閃斷而拋出異常中斷
    for submission in subreddit.stream.submissions(skip_existing=True):
        if "Agent" in submission.title:
            print(f"Found match: {submission.title}")
            reply_text = generate_reply(submission.selftext)
            # 沒有異常捕捉，API 限制會導致崩潰
            submission.reply(reply_text)

if __name__ == "__main__":
    monitor_and_reply()

開發者與 AI 的對話記錄

Ponytail (極簡主義者): 「看吧，20 行程式碼就搞定了 Reddit 監控與回覆。PRAW 庫 (Python Reddit API Wrapper) 已經幫我們封裝好 stream 方法，直接跑個無窮迴圈就好了，非常符合極簡哲學。」

Agent Reach (容錯專家): 「這在測試環境或許行得通，但在生產環境 (Production Environment) 中是個災難。Reddit 的 API 有嚴格的速率限制，且 stream.submissions() 在遇到暫時性網路中斷 (Transient Network Failure) 時會直接拋出 RequestException 並終止整個進程。」

Ponytail: 「那加個 try-except 包起來，遇到錯誤就 time.sleep(60) 不就好了？」

Agent Reach: 「不夠。我們需要實作：1) 指數退避策略 (Exponential Backoff Strategy) 處理 429 錯誤；2) 狀態持久化 (State Persistence) 記錄最後處理的貼文 ID，防止重啟時重複回覆；3) LLM 請求超時與重試機制。這才是真正的容錯設計。」

重構後的代碼：導入 Agent Reach 容錯機制

以下重構後的代碼保留了極簡的業務邏輯，但在網路通訊層加入了完善的安全防護與邊界檢查 (Boundary Checking)。

import praw
import time
import logging
from prawcore.exceptions import RequestException, ResponseException
from requests.exceptions import ConnectionError, Timeout
from llm_client import generate_reply

# 配置標準日誌 (Logging)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class RedditMonitorAgent:
    def __init__(self, subreddit_name: str, max_retries: int = 5):
        self.subreddit_name = subreddit_name
        self.max_retries = max_retries
        self.reddit = praw.Reddit(
            client_id="YOUR_CLIENT_ID",
            client_secret="YOUR_CLIENT_SECRET",
            user_agent="robust_monitor_agent/2.0",
            timeout=15 # 設定明確的超時時間
        )
        # 簡單的狀態持久化 (實際應用可改用 SQLite 或 Redis)
        self.processed_ids = set()

    def generate_and_post_reply(self, submission) -> bool:
        """生成並發送回覆，包含重試邏輯"""
        retries = 0
        while retries < self.max_retries:
            try:
                reply_text = generate_reply(submission.selftext)
                if not reply_text:
                    logger.warning(f"LLM 產生空回覆，略過貼文 {submission.id}")
                    return False

                submission.reply(reply_text)
                logger.info(f"成功回覆貼文: {submission.id}")
                return True

            except ResponseException as e:
                if e.response.status_code == 429:
                    wait_time = 2 ** retries * 10
                    logger.warning(f"觸發速率限制 (Rate Limit)。等待 {wait_time} 秒後重試...")
                    time.sleep(wait_time)
                    retries += 1
                else:
                    logger.error(f"Reddit API 回應錯誤: {e}")
                    break
            except Exception as e:
                logger.error(f"未預期的錯誤: {e}")
                break
                
        return False

    def start_monitoring(self):
        """啟動穩健的監控迴圈"""
        logger.info(f"開始監控 Subreddit: {self.subreddit_name}")
        while True:
            try:
                subreddit = self.reddit.subreddit(self.subreddit_name)
                for submission in subreddit.stream.submissions(skip_existing=True):
                    if submission.id in self.processed_ids:
                        continue
                        
                    if "Agent" in submission.title or "AI" in submission.title:
                        logger.info(f"發現匹配貼文: {submission.title} ({submission.id})")
                        success = self.generate_and_post_reply(submission)
                        if success:
                            self.processed_ids.add(submission.id)

            except (RequestException, ConnectionError, Timeout) as e:
                logger.error(f"網路連線異常: {e}。5 秒後重新建立連線...")
                time.sleep(5)
            except Exception as e:
                logger.critical(f"嚴重錯誤: {e}")
                time.sleep(30)

if __name__ == "__main__":
    agent = RedditMonitorAgent(subreddit_name="programming")
    agent.start_monitoring()

效益分析表格與解讀

解讀： 在此範例中，我們並未違背 Ponytail 的極簡設計，核心邏輯依然清晰易懂。但在網路連線與 API 呼叫的交界處，我們嚴格遵守 Agent Reach 的規範。透過指數退避、明確的超時設定以及冪等性設計，系統獲得了強大的自我修復 (Self-healing) 能力。在面對社群平台瞬息萬變的網路狀態時，這樣的設計是建構企業級 AI 代理的基石。