493 lines
18 KiB
Python
493 lines
18 KiB
Python
"""
|
||
Ollama Provider - Phase 24 ADR-052
|
||
====================================
|
||
本地 / 私有 LLM 推理 Provider。
|
||
|
||
搬移自: openclaw.py _call_ollama (L349-409)
|
||
特性: 免費、隱私安全 (local)、可依 ADR-110 指向 GCP-A/GCP-B/111。
|
||
|
||
2026-04-02 ogt: Phase 24-A 從 openclaw.py 抽出
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import time
|
||
|
||
import httpx
|
||
import structlog
|
||
|
||
from src.core.config import get_settings
|
||
from src.plugins.mcp.interfaces import MCPTool
|
||
from src.services.ai_providers.interfaces import (
|
||
AIResult,
|
||
is_provider_enabled_by_env,
|
||
)
|
||
from src.services.ai_providers.tool_schema import openai_tools_for_agent
|
||
from src.services.model_registry import get_model_registry
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
settings = get_settings()
|
||
|
||
_GCP_LIGHTWEIGHT_MODELS = {
|
||
"gemma3:4b",
|
||
}
|
||
|
||
|
||
def _normalized_url(value: str | None) -> str:
|
||
return (value or "").rstrip("/")
|
||
|
||
|
||
def _is_gcp_alert_lane(endpoint_url: str) -> bool:
|
||
"""Return true for the CPU-only GCP-A/B synchronous alert lane."""
|
||
endpoint = _normalized_url(endpoint_url)
|
||
return endpoint in {
|
||
_normalized_url(getattr(settings, "OLLAMA_URL", "")),
|
||
_normalized_url(getattr(settings, "OLLAMA_SECONDARY_URL", "")),
|
||
}
|
||
|
||
|
||
def _resolve_model_for_endpoint(
|
||
*,
|
||
requested_model: str,
|
||
endpoint_url: str,
|
||
context: dict | None,
|
||
) -> str:
|
||
"""
|
||
Keep non-diagnosis calls from polluting the GCP diagnosis lane.
|
||
|
||
GCP-A/B are allowed to run the deep incident diagnosis model because the
|
||
alert goal is correctness and resolution, not the fastest Telegram card.
|
||
Accidental non-diagnosis workloads still fall back to the lightweight health
|
||
model so embedding/Hermes/background calls cannot occupy the same lane.
|
||
"""
|
||
model_name = requested_model.strip()
|
||
context = context or {}
|
||
allow_gcp_heavy = bool(context.get("allow_gcp_heavy_model"))
|
||
task_type = str(context.get("task_type") or context.get("intent_hint") or "").lower()
|
||
is_deep_diagnosis = task_type in {"diagnose", "alert_deep", "incident_diagnosis"}
|
||
|
||
if (
|
||
_is_gcp_alert_lane(endpoint_url)
|
||
and not allow_gcp_heavy
|
||
and not is_deep_diagnosis
|
||
and model_name not in _GCP_LIGHTWEIGHT_MODELS
|
||
):
|
||
fallback_model = str(getattr(settings, "OLLAMA_HEALTH_CHECK_MODEL", "gemma3:4b")).strip() or "gemma3:4b"
|
||
logger.warning(
|
||
"ollama_gcp_non_diagnosis_model_coerced",
|
||
endpoint=endpoint_url,
|
||
requested_model=model_name,
|
||
safe_model=fallback_model,
|
||
task_type=task_type,
|
||
)
|
||
return fallback_model
|
||
|
||
return model_name
|
||
|
||
|
||
class OllamaProvider:
|
||
"""
|
||
Ollama 本地 LLM Provider
|
||
|
||
privacy_level: local (可處理機密資料)
|
||
capabilities: rca, chat, code_review
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self._http_client: httpx.AsyncClient | None = None
|
||
|
||
async def _get_client(self) -> httpx.AsyncClient:
|
||
if self._http_client is None or self._http_client.is_closed:
|
||
self._http_client = httpx.AsyncClient(
|
||
timeout=httpx.Timeout(120.0, connect=10.0),
|
||
)
|
||
return self._http_client
|
||
|
||
def _endpoint_url(self) -> str:
|
||
return settings.OLLAMA_URL
|
||
|
||
@property
|
||
def name(self) -> str:
|
||
return "ollama"
|
||
|
||
@property
|
||
def is_enabled(self) -> bool:
|
||
return is_provider_enabled_by_env("ollama")
|
||
|
||
@property
|
||
def capabilities(self) -> set[str]:
|
||
return {"rca", "chat", "code_review"}
|
||
|
||
@property
|
||
def privacy_level(self) -> str:
|
||
return "local"
|
||
|
||
async def analyze(
|
||
self,
|
||
prompt: str,
|
||
context: dict | None = None,
|
||
) -> AIResult:
|
||
start = time.perf_counter()
|
||
try:
|
||
client = await self._get_client()
|
||
|
||
registry = get_model_registry()
|
||
endpoint_url = self._endpoint_url()
|
||
requested_model = str((context or {}).get("ollama_model") or registry.get_model("ollama", "rca")).strip()
|
||
model_name = _resolve_model_for_endpoint(
|
||
requested_model=requested_model,
|
||
endpoint_url=endpoint_url,
|
||
context=context,
|
||
)
|
||
options = registry.get_provider_options("ollama")
|
||
|
||
# P0 2026-04-04 Claude Code: per-task timeout（Option C 分情境）
|
||
# FORCE_LOCAL/diagnose → OLLAMA_DIAGNOSE_TIMEOUT_SECONDS
|
||
# 其他 → OPENCLAW_TIMEOUT（既有設定）
|
||
task_type = (context or {}).get("task_type", "")
|
||
if task_type in ("diagnose", "force_local"):
|
||
read_timeout = float(getattr(settings, "OLLAMA_DIAGNOSE_TIMEOUT_SECONDS", 200))
|
||
else:
|
||
read_timeout = float(settings.OPENCLAW_TIMEOUT)
|
||
|
||
response = await client.post(
|
||
f"{endpoint_url}/api/generate",
|
||
json={
|
||
"model": model_name,
|
||
"prompt": prompt,
|
||
"stream": False,
|
||
"format": "json",
|
||
"options": {
|
||
"num_predict": options.get("num_predict", 1024),
|
||
"temperature": options.get("temperature", 0.1),
|
||
"top_p": options.get("top_p", 0.9),
|
||
},
|
||
},
|
||
timeout=httpx.Timeout(read_timeout, connect=10.0),
|
||
)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
result = data.get("response", "")
|
||
# I3 修復: 追蹤 tokens
|
||
tokens = data.get("eval_count", 0) + data.get("prompt_eval_count", 0)
|
||
latency = (time.perf_counter() - start) * 1000
|
||
|
||
logger.info(
|
||
"ollama_provider_success",
|
||
response_length=len(result),
|
||
tokens=tokens,
|
||
latency_ms=round(latency, 1),
|
||
model=model_name,
|
||
)
|
||
return AIResult(
|
||
raw_response=result,
|
||
success=True,
|
||
provider=self.name,
|
||
tokens=tokens,
|
||
latency_ms=latency,
|
||
)
|
||
|
||
except httpx.TimeoutException as e:
|
||
latency = (time.perf_counter() - start) * 1000
|
||
logger.warning("ollama_provider_timeout", error=str(e), latency_ms=round(latency, 1))
|
||
return AIResult(raw_response="", success=False, provider=self.name, latency_ms=latency, error=f"Timeout: {e}")
|
||
|
||
except Exception as e:
|
||
latency = (time.perf_counter() - start) * 1000
|
||
logger.warning("ollama_provider_failed", error=str(e), latency_ms=round(latency, 1))
|
||
return AIResult(raw_response="", success=False, provider=self.name, latency_ms=latency, error=str(e))
|
||
|
||
async def analyze_with_tools(
|
||
self,
|
||
prompt: str,
|
||
available_tools: list[MCPTool],
|
||
tool_executor,
|
||
max_iterations: int = 5,
|
||
agent_role: str = "openclaw",
|
||
context: dict | None = None,
|
||
) -> AIResult:
|
||
"""Run Ollama chat tool calling loop when the local model supports tools."""
|
||
|
||
if not available_tools:
|
||
return await self.analyze(prompt, context=context)
|
||
|
||
tools_schema = openai_tools_for_agent(available_tools, agent_role)
|
||
if not tools_schema:
|
||
return AIResult(
|
||
raw_response="",
|
||
success=False,
|
||
provider=self.name,
|
||
error=f"No MCP tools allowed for agent_role={agent_role}",
|
||
)
|
||
|
||
start = time.perf_counter()
|
||
total_tokens = 0
|
||
messages: list[dict] = [{"role": "user", "content": prompt}]
|
||
registry = get_model_registry()
|
||
model_name = str((context or {}).get("ollama_model") or registry.get_model("ollama", "rca")).strip()
|
||
options = registry.get_provider_options("ollama")
|
||
task_type = (context or {}).get("task_type", "")
|
||
if task_type in ("diagnose", "force_local"):
|
||
read_timeout = float(getattr(settings, "OLLAMA_DIAGNOSE_TIMEOUT_SECONDS", 200))
|
||
else:
|
||
read_timeout = float(settings.OPENCLAW_TIMEOUT)
|
||
|
||
try:
|
||
client = await self._get_client()
|
||
last_content = ""
|
||
for iteration in range(max_iterations):
|
||
response = await client.post(
|
||
f"{self._endpoint_url()}/api/chat",
|
||
json={
|
||
"model": model_name,
|
||
"messages": messages,
|
||
"stream": False,
|
||
"tools": tools_schema,
|
||
"options": {
|
||
"num_predict": options.get("num_predict", 1024),
|
||
"temperature": options.get("temperature", 0.1),
|
||
"top_p": options.get("top_p", 0.9),
|
||
},
|
||
},
|
||
timeout=httpx.Timeout(read_timeout, connect=10.0),
|
||
)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
total_tokens += int(data.get("eval_count", 0) or 0)
|
||
total_tokens += int(data.get("prompt_eval_count", 0) or 0)
|
||
|
||
message = data.get("message") or {}
|
||
last_content = message.get("content") or last_content
|
||
tool_calls = message.get("tool_calls") or []
|
||
if not tool_calls:
|
||
latency = (time.perf_counter() - start) * 1000
|
||
return AIResult(
|
||
raw_response=last_content or json.dumps(data, ensure_ascii=False),
|
||
success=True,
|
||
provider=f"{self.name}_agent_loop",
|
||
tokens=total_tokens,
|
||
latency_ms=latency,
|
||
)
|
||
|
||
messages.append(message)
|
||
for call in tool_calls:
|
||
function = call.get("function") or {}
|
||
tool_name = function.get("name", "")
|
||
arguments = function.get("arguments") or {}
|
||
result = await tool_executor(tool_name, arguments)
|
||
messages.append({
|
||
"role": "tool",
|
||
"content": json.dumps(
|
||
result.to_dict() if hasattr(result, "to_dict") else result,
|
||
ensure_ascii=False,
|
||
default=str,
|
||
),
|
||
})
|
||
|
||
logger.debug(
|
||
"ollama_agent_loop_iteration",
|
||
provider=self.name,
|
||
agent_role=agent_role,
|
||
iteration=iteration + 1,
|
||
tool_calls=len(tool_calls),
|
||
)
|
||
|
||
latency = (time.perf_counter() - start) * 1000
|
||
return AIResult(
|
||
raw_response=last_content,
|
||
success=False,
|
||
provider=f"{self.name}_agent_loop",
|
||
tokens=total_tokens,
|
||
latency_ms=latency,
|
||
error=f"Agent loop exceeded max_iterations={max_iterations}",
|
||
)
|
||
|
||
except Exception as e:
|
||
latency = (time.perf_counter() - start) * 1000
|
||
logger.warning(
|
||
"ollama_agent_loop_failed",
|
||
provider=self.name,
|
||
agent_role=agent_role,
|
||
error=str(e),
|
||
latency_ms=round(latency, 1),
|
||
)
|
||
return AIResult(
|
||
raw_response="",
|
||
success=False,
|
||
provider=f"{self.name}_agent_loop",
|
||
tokens=total_tokens,
|
||
latency_ms=latency,
|
||
error=str(e),
|
||
)
|
||
|
||
async def health_check(self) -> bool:
|
||
try:
|
||
client = await self._get_client()
|
||
resp = await client.get(f"{self._endpoint_url()}/api/tags", timeout=5.0)
|
||
return resp.status_code == 200
|
||
except Exception:
|
||
return False
|
||
|
||
async def close(self) -> None:
|
||
if self._http_client:
|
||
await self._http_client.aclose()
|
||
self._http_client = None
|
||
|
||
|
||
# 2026-05-06 Codex — 188 不再作為 Ollama Provider；本地備援統一命名為 ollama_local。
|
||
class OllamaLocalProvider(OllamaProvider):
|
||
"""
|
||
Ollama Local fallback Provider
|
||
|
||
使用 OLLAMA_FALLBACK_URL 作為本地最後防線端點。
|
||
ADR-110 目前設定為 110 nginx proxy → 111 Ollama；188 不得再作為 Ollama provider。
|
||
"""
|
||
|
||
@property
|
||
def name(self) -> str:
|
||
return "ollama_local"
|
||
|
||
@property
|
||
def is_enabled(self) -> bool:
|
||
import os
|
||
# 優先查 ENABLE_OLLAMA_LOCAL；若未設定（預設 true）則看 OLLAMA_FALLBACK_URL 是否有值。
|
||
env_override = os.getenv("ENABLE_OLLAMA_LOCAL", "true").lower() == "true"
|
||
if not env_override:
|
||
return False
|
||
# OLLAMA_FALLBACK_URL 空字串 → 未設定本地節點 → 停用。
|
||
return bool(getattr(settings, "OLLAMA_FALLBACK_URL", ""))
|
||
|
||
def _endpoint_url(self) -> str:
|
||
return getattr(settings, "OLLAMA_FALLBACK_URL", "")
|
||
|
||
async def analyze(
|
||
self,
|
||
prompt: str,
|
||
context: dict | None = None,
|
||
) -> AIResult:
|
||
start = time.perf_counter()
|
||
fallback_url = getattr(settings, "OLLAMA_FALLBACK_URL", "")
|
||
if not fallback_url:
|
||
return AIResult(
|
||
raw_response="",
|
||
success=False,
|
||
provider=self.name,
|
||
error="OLLAMA_FALLBACK_URL not configured",
|
||
)
|
||
|
||
try:
|
||
client = await self._get_client()
|
||
|
||
registry = get_model_registry()
|
||
# 嘗試取本地 fallback 專屬設定，fallback 到 ollama 預設。
|
||
try:
|
||
model_name = str((context or {}).get("ollama_model") or registry.get_model("ollama_local", "rca")).strip()
|
||
except Exception:
|
||
model_name = str((context or {}).get("ollama_model") or getattr(settings, "OLLAMA_HEALTH_CHECK_MODEL", "qwen2.5:7b-instruct")).strip()
|
||
|
||
try:
|
||
options = registry.get_provider_options("ollama_local")
|
||
except Exception:
|
||
options = registry.get_provider_options("ollama")
|
||
|
||
# 本地備援：固定使用較長 timeout，避免 111 模型載入時被過早判死。
|
||
task_type = (context or {}).get("task_type", "")
|
||
if task_type in ("diagnose", "force_local"):
|
||
read_timeout = float(getattr(settings, "OLLAMA_DIAGNOSE_TIMEOUT_SECONDS", 200))
|
||
else:
|
||
read_timeout = float(settings.OPENCLAW_TIMEOUT)
|
||
|
||
response = await client.post(
|
||
f"{fallback_url}/api/generate",
|
||
json={
|
||
"model": model_name,
|
||
"prompt": prompt,
|
||
"stream": False,
|
||
"format": "json",
|
||
"options": {
|
||
"num_predict": options.get("num_predict", 1024),
|
||
"temperature": options.get("temperature", 0.1),
|
||
"top_p": options.get("top_p", 0.9),
|
||
},
|
||
},
|
||
timeout=httpx.Timeout(read_timeout, connect=10.0),
|
||
)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
result = data.get("response", "")
|
||
tokens = data.get("eval_count", 0) + data.get("prompt_eval_count", 0)
|
||
latency = (time.perf_counter() - start) * 1000
|
||
|
||
logger.info(
|
||
"ollama_local_provider_success",
|
||
response_length=len(result),
|
||
tokens=tokens,
|
||
latency_ms=round(latency, 1),
|
||
endpoint=fallback_url,
|
||
model=model_name,
|
||
)
|
||
return AIResult(
|
||
raw_response=result,
|
||
success=True,
|
||
provider=self.name,
|
||
tokens=tokens,
|
||
latency_ms=latency,
|
||
)
|
||
|
||
except httpx.TimeoutException as e:
|
||
latency = (time.perf_counter() - start) * 1000
|
||
logger.warning("ollama_local_provider_timeout", error=str(e), latency_ms=round(latency, 1))
|
||
return AIResult(raw_response="", success=False, provider=self.name, latency_ms=latency, error=f"Timeout: {e}")
|
||
|
||
except Exception as e:
|
||
latency = (time.perf_counter() - start) * 1000
|
||
logger.warning("ollama_local_provider_failed", error=str(e), latency_ms=round(latency, 1))
|
||
return AIResult(raw_response="", success=False, provider=self.name, latency_ms=latency, error=str(e))
|
||
|
||
async def health_check(self) -> bool:
|
||
fallback_url = getattr(settings, "OLLAMA_FALLBACK_URL", "")
|
||
if not fallback_url:
|
||
return False
|
||
try:
|
||
client = await self._get_client()
|
||
resp = await client.get(f"{fallback_url}/api/tags", timeout=5.0)
|
||
return resp.status_code == 200
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
class OllamaGcpBProvider(OllamaProvider):
|
||
"""
|
||
GCP-B Secondary Ollama Provider
|
||
|
||
繼承 OllamaProvider，使用 OLLAMA_SECONDARY_URL（34.21.145.224:11434）。
|
||
ADR-110 三層容災：GCP-A → GCP-B → Local(111)。
|
||
OllamaFailoverManager 回傳 provider_name="ollama_gcp_b" 時由此 Provider 執行。
|
||
|
||
2026-05-04 ogt + Claude Sonnet 4.6: ADR-110 GCP-B 容災補全
|
||
根因：AIProviderRegistry 缺少 "ollama_gcp_b" → not_registered → 跳 Gemini
|
||
"""
|
||
|
||
@property
|
||
def name(self) -> str:
|
||
return "ollama_gcp_b"
|
||
|
||
@property
|
||
def is_enabled(self) -> bool:
|
||
return bool(getattr(settings, "OLLAMA_SECONDARY_URL", ""))
|
||
|
||
def _endpoint_url(self) -> str:
|
||
return getattr(settings, "OLLAMA_SECONDARY_URL", "")
|
||
|
||
async def health_check(self) -> bool:
|
||
url = getattr(settings, "OLLAMA_SECONDARY_URL", "")
|
||
if not url:
|
||
return False
|
||
try:
|
||
client = await self._get_client()
|
||
resp = await client.get(f"{url}/api/tags", timeout=5.0)
|
||
return resp.status_code == 200
|
||
except Exception:
|
||
return False
|