118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
"""
|
||
AWOOOI Chat Manager - 統帥對話核心
|
||
===================================
|
||
Phase 21.5: 實作 Telegram 互動對話功能
|
||
|
||
職責:
|
||
1. 整合系統上下文 (K3s 狀態, 最近告警, 目前時間)
|
||
2. 決定對話風格 (OpenClaw 專業風 vs Nemo 參謀風)
|
||
3. 調用 LLM (Nemo-4B / Gemini) 產出回應
|
||
4. 遵守 SOUL.md Nothing.tech 純淨美學
|
||
|
||
2026-03-31 ogt: 初版建立
|
||
"""
|
||
|
||
import structlog
|
||
from datetime import datetime
|
||
from src.utils.timezone import now_taipei
|
||
from src.services.nvidia_provider import get_nvidia_provider
|
||
from src.repositories.k8s_repository import get_k8s_repository
|
||
from src.repositories.incident_repository import get_incident_repository
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
class ChatManager:
|
||
"""
|
||
AWOOOI 對話管理器 - 系統的大腦與聲帶
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.nvidia = get_nvidia_provider()
|
||
self.k8s = get_k8s_repository()
|
||
self.incidents = get_incident_repository()
|
||
|
||
async def get_system_context(self) -> str:
|
||
"""
|
||
收集系統即時上下文，供 LLM 參考
|
||
"""
|
||
now = now_taipei()
|
||
|
||
# 1. K3s 狀態
|
||
k8s_status = await self.k8s.get_pod_status_summary(namespace="awoooi-prod")
|
||
cluster_info = f"Cluster: {k8s_status['running']}/{k8s_status['total']} Pods Running"
|
||
if k8s_status['problem_pods']:
|
||
cluster_info += f", {len(k8s_status['problem_pods'])} anomalies detected."
|
||
|
||
# 2. 最近告警 (取 3 筆)
|
||
active_incidents = await self.incidents.get_active()
|
||
incident_summary = "None"
|
||
if active_incidents:
|
||
lines = []
|
||
for inc in active_incidents[:3]:
|
||
lines.append(f"- {inc.incident_id}: {inc.status.value} (Severity: {inc.severity.value})")
|
||
incident_summary = "\n".join(lines)
|
||
|
||
context = f"""
|
||
## Current System Context (Taipei Time: {now.strftime('%Y-%m-%d %H:%M:%S')})
|
||
- Environment: AWOOOI Production (K3s)
|
||
- {cluster_info}
|
||
- Active Incidents:
|
||
{incident_summary}
|
||
"""
|
||
return context
|
||
|
||
async def generate_response(
|
||
self,
|
||
user_id: int,
|
||
username: str,
|
||
message_text: str
|
||
) -> str:
|
||
"""
|
||
根據統帥訊息產生回覆
|
||
"""
|
||
system_context = await self.get_system_context()
|
||
|
||
# 判定是否在跟 Nemo 對話
|
||
is_asking_nemo = "nemo" in message_text.lower()
|
||
|
||
role_description = "You are OpenClaw, the AI operations assistant for AWOOOI platform."
|
||
if is_asking_nemo:
|
||
role_description = "You are Nemo-4B, the elite AI tactical advisor for AWOOOI. Address the user as 'Supreme Commander' (統帥)."
|
||
|
||
system_prompt = f"""{role_description}
|
||
{system_context}
|
||
|
||
## Guidelines:
|
||
1. Keep responses extremely concise and professional (Nothing.tech aesthetic).
|
||
2. For status queries, provide precise data.
|
||
3. For general chat, be supportive but focused on operations.
|
||
4. Language: Preferred Traditional Chinese (繁體中文).
|
||
5. No emojis except for functional ones (🚨, ✅, 📊).
|
||
"""
|
||
|
||
try:
|
||
# 優先使用 NVIDIA Nemo-4B
|
||
response, success, tokens, cost = await self.nvidia.chat(
|
||
prompt=f"{system_prompt}\n\nCommander's Message: {message_text}",
|
||
model="nvidia/nemotron-mini-4b-instruct",
|
||
max_tokens=1024
|
||
)
|
||
|
||
if not success:
|
||
return "🛸 抱歉統帥，Nemo 參謀暫時離線。請稍後再試。"
|
||
|
||
return response.strip()
|
||
|
||
except Exception as e:
|
||
logger.exception("chat_generation_error", error=str(e))
|
||
return "⚠️ 通訊鏈路異常，無法聯繫 AI 腦區。"
|
||
|
||
# Singleton
|
||
_chat_manager = None
|
||
|
||
def get_chat_manager() -> ChatManager:
|
||
global _chat_manager
|
||
if _chat_manager is None:
|
||
_chat_manager = ChatManager()
|
||
return _chat_manager
|