import requests import json # 雖然 requests 會處理 json，但保留導入並無害 import time from datetime import datetime import difflib # 設定 Ollama API 的基礎 URL OLLAMA_BASE_URL = "http://localhost:11434/api" OLLAMA_GENERATE_URL = f"{OLLAMA_BASE_URL}/generate" OLLAMA_TAGS_URL = f"{OLLAMA_BASE_URL}/tags" # 用於獲取模型列表的 API 端點 # 不再需要硬性指定模型列表 # MODELS = ["TaiwanPro-r", "gemmapro", "gemmapro-r", "gemmapro-20kctx"] def get_available_models(): """向 Ollama API 查詢本機可用的模型列表""" print("[查詢] 正在獲取本機可用的 Ollama 模型...") try: response = requests.get(OLLAMA_TAGS_URL) response.raise_for_status() # 檢查 HTTP 請求是否成功 data = response.json() # 從回應中提取模型名稱列表 # API 回應格式通常是 {"models": [{"name": "model1:tag", ...}, ...]} if "models" in data and isinstance(data["models"], list): model_names = [model.get("name") for model in data["models"] if model.get("name")] if not model_names: print("[警告] 未在本機找到任何 Ollama 模型。請確認已安裝並載入模型。") return [] print(f"[成功] 找到可用的模型: {', '.join(model_names)}") return model_names else: print("[錯誤] 無法從 Ollama API 回應中解析模型列表。回應格式可能已變更。") print(f"原始回應: {data}") return [] except requests.exceptions.ConnectionError: print(f"[錯誤] 無法連接到 Ollama 服務於 {OLLAMA_BASE_URL}。請確認 Ollama 正在運行。") return None # 返回 None 表示連接失敗 except requests.exceptions.RequestException as e: print(f"[錯誤] 查詢可用模型時發生錯誤: {e}") return [] except json.JSONDecodeError: print(f"[錯誤] 無法解析來自 {OLLAMA_TAGS_URL} 的回應 (非 JSON 格式): {response.text}") return [] def send_request_to_ollama(prompt, model): """向指定模型發送請求並獲取回應""" data = { "model": model, "prompt": prompt, "stream": False } try: # 使用定義好的 generate URL response = requests.post(OLLAMA_GENERATE_URL, json=data) response.raise_for_status() # 檢查 HTTP 請求是否成功 (狀態碼 2xx) return response.json()["response"] except requests.exceptions.RequestException as e: print(f"[錯誤] 模型 {model} 請求失敗: {e}") # 如果是因為模型不存在 (404 Not Found)，提供更具體的提示 if response is not None and response.status_code == 404: print(f" 提示：模型 '{model}' 可能未完全下載或不存在於 Ollama 中。") return f"[錯誤] 模型 '{model}' 未找到或不可用。" return f"[錯誤] 向 {model} 發送請求時發生錯誤: {str(e)}" # 提供更明確的錯誤訊息 except KeyError: print(f"[錯誤] 模型 {model} 回應格式不符預期，找不到 'response' 鍵。") return f"[錯誤] 模型 {model} 回應格式錯誤。" except json.JSONDecodeError: print(f"[錯誤] 模型 {model} 回應非有效的 JSON 格式: {response.text}") return f"[錯誤] 無法解析來自 {model} 的回應。" def initialize_markdown_file(models_to_run): """初始化 Markdown 報告檔案，包含 YAML metadata""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") metadata = { "title": "多模型推理彙整報告 (自動偵測可用模型)", "date": timestamp, "models_queried": models_to_run, # 使用實際查詢的模型列表 "author": "自動化程式", "description": "本報告整合多個本機可用的模型對多個問題的回應，進行去蕪存菁後的彙整。" } try: # 增加檔案操作的錯誤處理 with open("output_moremodels_auto.md", "w", encoding="utf-8") as file: # 更改檔案名以區分 file.write("---\n") for key, value in metadata.items(): if isinstance(value, list): file.write(f"{key}:\n") for item in value: file.write(f" - {item}\n") # 標準 YAML 列表縮排 else: file.write(f"{key}: {value}\n") file.write("---\n\n") file.write(f"# {metadata['title']}\n\n") file.write(f"產出時間: {timestamp}\n\n") if models_to_run: # 只有在找到模型時才寫入 file.write(f"使用模型: {', '.join(models_to_run)}\n\n---\n\n") else: file.write("未找到可用的模型進行查詢。\n\n---\n\n") print("[初始化] 已建立 output_moremodels_auto.md") return True except IOError as e: print(f"[錯誤] 無法寫入檔案 output_moremodels_auto.md: {e}") return False # 返回 False 表示初始化失敗 def append_to_markdown(index, prompt, responses): """將問題與各模型回應結果寫入 Markdown 檔案""" try: # 增加檔案操作的錯誤處理 with open("output_moremodels_auto.md", "a", encoding="utf-8") as file: file.write(f"## 問題 {index}\n\n") file.write(f"### 提問\n\n```\n{prompt}\n```\n\n") if not responses: file.write("沒有任何模型提供回應。\n\n") else: for model, response in responses.items(): # 確保 response 是字串，避免後續處理錯誤 response_text = str(response) if response is not None else "[無回應]" file.write(f"### 模型：{model}\n\n{response_text.strip()}\n\n") # 自動生成摘要彙整 # 確保將有效的回應傳遞給摘要函數 valid_responses = {m: r for m, r in responses.items() if isinstance(r, str) and not r.startswith("[錯誤]")} summary = summarize_responses(prompt, valid_responses) # 傳遞有效的回應字典 file.write(f"### 彙整摘要\n\n{summary}\n\n") file.write("---\n\n") except IOError as e: print(f"[錯誤] 無法附加內容至檔案 output_moremodels_auto.md: {e}") def summarize_responses(prompt, responses): """ 將多個模型的回應進行比較，提取相似的句子，並整合成通順的摘要。 (此函數邏輯保持不變) """ # 如果沒有有效的回應，直接返回提示訊息 if not responses: return "沒有從任何模型收到有效回應可供摘要。" # 將每個回應分句 sentence_lists = [] model_names = list(responses.keys()) # 記錄模型名稱順序 for response in responses.values(): # 更穩健的分句方式，處理不同結尾符號和換行 processed_response = response.replace('\n', ' ') sentences = [] current_sentence = "" for char in processed_response: current_sentence += char if char in ['。', '！', '？', '.', '!', '?']: trimmed_sentence = current_sentence.strip() if trimmed_sentence: sentences.append(trimmed_sentence) current_sentence = "" trimmed_sentence = current_sentence.strip() if trimmed_sentence: sentences.append(trimmed_sentence) sentence_lists.append(sentences) processed_indices = set() summary_sentences = [] # 將模型索引與句子索引和句子本身綁定 all_sentences = [(i, j, sent) for i, lst in enumerate(sentence_lists) for j, sent in enumerate(lst)] for idx1 in range(len(all_sentences)): model_idx1, sent_idx1, sent1 = all_sentences[idx1] if (model_idx1, sent_idx1) in processed_indices: continue best_match = None max_similarity = 0.7 for idx2 in range(idx1 + 1, len(all_sentences)): model_idx2, sent_idx2, sent2 = all_sentences[idx2] if model_idx1 == model_idx2: # 確保是不同模型的回應 continue if (model_idx2, sent_idx2) in processed_indices: continue similarity = difflib.SequenceMatcher(None, sent1, sent2).ratio() if similarity > max_similarity: max_similarity = similarity chosen_sentence = sent1 if len(sent1) <= len(sent2) else sent2 best_match = ((model_idx1, sent_idx1), (model_idx2, sent_idx2), chosen_sentence) if best_match: idx_pair1, idx_pair2, chosen = best_match processed_indices.add(idx_pair1) processed_indices.add(idx_pair2) summary_sentences.append(chosen) if not summary_sentences: # 如果沒有找到足夠相似的句子，可以選擇提供各模型的簡短預覽 summary = "各模型提供了不同的觀點，未偵測到足夠相似的核心內容可供直接彙整。重點預覽如下：\n\n" for model_index, model_name in enumerate(model_names): response_text = responses[model_name] preview = response_text.strip().replace('\n', ' ')[:100] # 截取前 100 字元預覽 summary += f"- **{model_name}**: {preview}...\n" else: summary = "綜合各模型的相似觀點，摘要如下：\n\n" unique_summary_sentences = [] for sentence in summary_sentences: if sentence not in unique_summary_sentences: unique_summary_sentences.append(sentence) for sentence in unique_summary_sentences: summary += f"- {sentence}\n" return summary def main(): # 1. 獲取可用的模型 available_models = get_available_models() # 如果無法連接到 Ollama 或找不到模型，則終止程式 if available_models is None: # 連接失敗 print("[終止] 無法執行，因無法連接到 Ollama 服務。") return if not available_models: # 找到服務但沒有模型 print("[終止] 無法執行，因未找到任何可用的模型。") # 仍嘗試初始化文件以記錄情況 initialize_markdown_file([]) return # 可自訂多個問題 questions = [ "介紹台灣的夜市文化", "台灣人工智慧發展的現況與挑戰為何？", "請用繁體中文解釋什麼是大型語言模型 (LLM)。" ] # 2. 初始化 Markdown 文件 (使用找到的模型列表) if not initialize_markdown_file(available_models): print("[終止] Markdown 檔案初始化失敗，程式結束。") return # 如果檔案無法建立，後續寫入會失敗 print("[開始] 向偵測到的模型發送請求...") # 3. 迭代問題和可用的模型 for i, prompt in enumerate(questions, 1): print(f"[處理中] 問題 {i}/{len(questions)}: {prompt[:30]}...") model_responses = {} for model in available_models: # 使用動態獲取的模型列表 print(f" └▶ 模型 {model} 推理中...") start_time = time.time() # 記錄開始時間 response = send_request_to_ollama(prompt, model) end_time = time.time() # 記錄結束時間 elapsed_time = end_time - start_time print(f" 回應耗時: {elapsed_time:.2f} 秒") # 顯示每個模型的回應時間 model_responses[model] = response # 如果需要，可以在這裡加回 time.sleep(1) # 4. 將結果寫入 Markdown append_to_markdown(i, prompt, model_responses) print(f"[完成] 問題 {i} 已處理並寫入檔案。") print(f"[完成] 所有問題已處理完畢，結果保存在 output_moremodels_auto.md") if __name__ == "__main__": main()