moremodels / moremodels2.py

Upload moremodels2.py

d8a89db verified over 1 year ago

12 kB

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()

	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()