ollama_client.py
· 3.1 KiB · Python
原始文件
import requests
class OllamaClient:
def __init__(self, api_url, api_key=None, timeout=30):
"""
初始化 OllamaClient。
:param api_url: API 基本端點,例如 'http://192.168.88.82:11434'
:param api_key: 如果需要身份驗證,填入 API 金鑰;否則為 None。
:param timeout: 請求超時時間(秒)。
"""
self.api_url = api_url.rstrip("/") # 移除末尾的斜槓以避免拼接錯誤
self.api_key = api_key
self.timeout = timeout
self.headers = {"Content-Type": "application/json"}
if self.api_key:
self.headers["Authorization"] = f"Bearer {self.api_key}"
def process_text(self, text, model="qwen2.5:14b", additional_prompt=None):
"""
發送文字到指定模型,並獲取處理後的結果。
:param text: 要處理的文字。
:param model: 指定要使用的模型(預設為 'qwen2.5:14b')。
:param additional_prompt: 可選的提示詞,將添加在文字之前。
:return: 處理後的文字。
"""
# 合成最終的 prompt
final_prompt = f"{additional_prompt}\n{text}" if additional_prompt else text
payload = {
"model": model,
"prompt": final_prompt,
"stream": False, # 確保回應為全量模式
}
try:
response = requests.post(
f"{self.api_url}/api/generate",
json=payload,
headers=self.headers,
timeout=self.timeout,
)
response.raise_for_status()
# 確保回應是 JSON 格式
try:
data = response.json()
except ValueError as e:
print(f"無法解析 Ollama LLM 的回應,非 JSON 格式: {e}")
print(f"回應內容: {response.text}")
return text
# 提取回應中的處理結果
processed_text = data.get("response")
if not processed_text:
print("Ollama LLM 回應中缺少 'response' 欄位。")
return text
return processed_text
except requests.exceptions.RequestException as e:
print(f"Ollama LLM 請求錯誤: {e}")
return text
def get_models(self):
"""
獲取目前可用的模型完整資訊。
:return: 包含模型完整資訊的列表(如果失敗,返回空列表)。
"""
try:
response = requests.get(
f"{self.api_url}/api/tags", headers=self.headers, timeout=self.timeout
)
response.raise_for_status()
data = response.json()
models = data.get("models", [])
if not models:
print("未能從 Ollama LLM 獲取模型列表。")
return []
return models
except requests.exceptions.RequestException as e:
print(f"Ollama LLM 請求錯誤: {e}")
return []
except ValueError:
print("無法解析 Ollama LLM 的回應。")
return []
| 1 | import requests |
| 2 | |
| 3 | |
| 4 | class OllamaClient: |
| 5 | def __init__(self, api_url, api_key=None, timeout=30): |
| 6 | """ |
| 7 | 初始化 OllamaClient。 |
| 8 | |
| 9 | :param api_url: API 基本端點,例如 'http://192.168.88.82:11434' |
| 10 | :param api_key: 如果需要身份驗證,填入 API 金鑰;否則為 None。 |
| 11 | :param timeout: 請求超時時間(秒)。 |
| 12 | """ |
| 13 | self.api_url = api_url.rstrip("/") # 移除末尾的斜槓以避免拼接錯誤 |
| 14 | self.api_key = api_key |
| 15 | self.timeout = timeout |
| 16 | self.headers = {"Content-Type": "application/json"} |
| 17 | if self.api_key: |
| 18 | self.headers["Authorization"] = f"Bearer {self.api_key}" |
| 19 | |
| 20 | def process_text(self, text, model="qwen2.5:14b", additional_prompt=None): |
| 21 | """ |
| 22 | 發送文字到指定模型,並獲取處理後的結果。 |
| 23 | |
| 24 | :param text: 要處理的文字。 |
| 25 | :param model: 指定要使用的模型(預設為 'qwen2.5:14b')。 |
| 26 | :param additional_prompt: 可選的提示詞,將添加在文字之前。 |
| 27 | :return: 處理後的文字。 |
| 28 | """ |
| 29 | # 合成最終的 prompt |
| 30 | final_prompt = f"{additional_prompt}\n{text}" if additional_prompt else text |
| 31 | |
| 32 | payload = { |
| 33 | "model": model, |
| 34 | "prompt": final_prompt, |
| 35 | "stream": False, # 確保回應為全量模式 |
| 36 | } |
| 37 | try: |
| 38 | response = requests.post( |
| 39 | f"{self.api_url}/api/generate", |
| 40 | json=payload, |
| 41 | headers=self.headers, |
| 42 | timeout=self.timeout, |
| 43 | ) |
| 44 | response.raise_for_status() |
| 45 | |
| 46 | # 確保回應是 JSON 格式 |
| 47 | try: |
| 48 | data = response.json() |
| 49 | except ValueError as e: |
| 50 | print(f"無法解析 Ollama LLM 的回應,非 JSON 格式: {e}") |
| 51 | print(f"回應內容: {response.text}") |
| 52 | return text |
| 53 | |
| 54 | # 提取回應中的處理結果 |
| 55 | processed_text = data.get("response") |
| 56 | if not processed_text: |
| 57 | print("Ollama LLM 回應中缺少 'response' 欄位。") |
| 58 | return text |
| 59 | return processed_text |
| 60 | except requests.exceptions.RequestException as e: |
| 61 | print(f"Ollama LLM 請求錯誤: {e}") |
| 62 | return text |
| 63 | |
| 64 | def get_models(self): |
| 65 | """ |
| 66 | 獲取目前可用的模型完整資訊。 |
| 67 | |
| 68 | :return: 包含模型完整資訊的列表(如果失敗,返回空列表)。 |
| 69 | """ |
| 70 | try: |
| 71 | response = requests.get( |
| 72 | f"{self.api_url}/api/tags", headers=self.headers, timeout=self.timeout |
| 73 | ) |
| 74 | response.raise_for_status() |
| 75 | data = response.json() |
| 76 | models = data.get("models", []) |
| 77 | if not models: |
| 78 | print("未能從 Ollama LLM 獲取模型列表。") |
| 79 | return [] |
| 80 | return models |
| 81 | except requests.exceptions.RequestException as e: |
| 82 | print(f"Ollama LLM 請求錯誤: {e}") |
| 83 | return [] |
| 84 | except ValueError: |
| 85 | print("無法解析 Ollama LLM 的回應。") |
| 86 | return [] |
| 87 |