apps/whatsapp_control_api/app/services/chat.py
text
from __future__ import annotations
import logging
from datetime import datetime
import httpx
from ..core.config import get_settings
from .web_search import WebSearchService, format_search_context, should_use_web_search
logger = logging.getLogger(__name__)
REMOTE_CHAT_PROVIDERS = {"openrouter", "openai", "openai_compatible"}
class AssistantChatService:
def __init__(self, web_search: WebSearchService | None = None) -> None:
self.web_search = web_search or WebSearchService()
@property
def settings(self):
return get_settings()
@property
def is_enabled(self) -> bool:
return self.settings.chat_enabled
async def answer(self, message: str, session_summary: str | None = None) -> str:
if is_hermes_agent_creation_question(message):
return hermes_agent_creation_reply()
search_results = []
if should_use_web_search(message):
search_results = await self.web_search.search(message)
if not self.is_enabled:
if search_results:
return web_search_sources_reply(search_results)
return local_assistant_reply(message)
provider = self.settings.chat_provider
if provider not in REMOTE_CHAT_PROVIDERS:
if search_results:
return web_search_sources_reply(search_results)
return local_assistant_reply(message)
if not self.settings.chat_api_key or not self.settings.chat_model:
if search_results:
return web_search_sources_reply(search_results)
return local_assistant_reply(message)
system_prompt = (
"Kamu adalah Hermes via WhatsApp. Jawab seperti AI asisten umum: natural, ringkas, "
"jelas, dan praktis dalam bahasa Indonesia. Jika user bertanya hal umum, jawab langsung. "
"Jika user meminta pekerjaan teknis yang berisiko, jelaskan bahwa akan diminta approval. "
"Jika ada konteks session, gunakan itu tanpa mengarang fakta. Jangan menutup jawaban dengan "
"kalimat bantuan generik kecuali memang dibutuhkan. Jika disediakan sumber web, gunakan sumber "
"itu sebagai dasar utama, jangan mengarang data di luar sumber, dan sertakan bagian 'Sumber' "
"berisi URL yang relevan."
)
if search_results:
system_prompt += (
f"\n\nTanggal runtime: {datetime.now().date().isoformat()}."
"\nKonteks web realtime:\n"
f"{format_search_context(search_results)}"
)
if session_summary:
system_prompt += f"\n\nKonteks singkat sesi:\n{session_summary}"
payload = {
"model": self.settings.chat_model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": message},
],
"temperature": self.settings.chat_temperature,
"max_tokens": self.settings.chat_max_tokens,
}
headers = {
"Authorization": f"Bearer {self.settings.chat_api_key}",
"Content-Type": "application/json",
}
try:
async with httpx.AsyncClient(timeout=45.0) as client:
response = await client.post(
f"{self.settings.chat_base_url}/chat/completions",
json=payload,
headers=headers,
)
response.raise_for_status()
except httpx.HTTPError as exc:
logger.warning("Chat provider request failed provider=%s model=%s error=%s", provider, self.settings.chat_model, exc)
if search_results:
return web_search_sources_reply(search_results)
return local_assistant_reply(message)
data = response.json()
content = data["choices"][0]["message"]["content"]
clean_content = str(content).strip()
return clean_content if clean_content else local_assistant_reply(message)
def local_assistant_reply(message: str) -> str:
lowered = message.strip().lower()
if lowered in {"halo", "hallo", "hai", "haii", "hi", "hello", "pagi", "siang", "sore", "malam"}:
return "Hai. Ada yang bisa saya bantu?"
if "bisa apa" in lowered or lowered in {"help", "menu", "panduan"}:
return (
"Saya bisa bantu chat umum, jawab pertanyaan, membaca konteks project, dan menjalankan task Hermes "
"kalau kamu kasih instruksi kerja yang jelas."
)
if any(word in lowered for word in {"siapa kamu", "kamu apa", "fungsi", "apa ini"}):
return (
"Saya Hermes via WhatsApp. Saya bisa dipakai seperti AI biasa untuk tanya-jawab, dan juga bisa "
"menerima instruksi kerja untuk dijalankan Hermes Agent di VPS."
)
if is_hermes_agent_creation_question(message):
return hermes_agent_creation_reply()
if lowered.endswith("?") or lowered.startswith(("apa ", "apakah ", "siapa ", "kapan ", "kenapa ", "mengapa ", "bagaimana ", "gimana ", "cara ")):
return (
"Saya bisa jawab, tapi mode chat AI eksternal belum aktif di service ini. "
"Untuk jawaban faktual yang lebih lengkap, aktifkan CHAT_PROVIDER, CHAT_MODEL, dan CHAT_API_KEY, "
"atau minta Hermes mencarinya dengan instruksi seperti: 'carikan informasi tentang ...'."
)
return "Saya paham. Kalau ada yang ingin ditanyakan atau dikerjakan, kirim langsung saja."
def web_search_sources_reply(results) -> str:
lines = [
"Saya menemukan beberapa sumber web terbaru, tetapi mode chat AI belum aktif untuk menyusun ringkasannya.",
"Sumber:",
]
for index, result in enumerate(results[:5], start=1):
lines.append(f"{index}. {result.title}\n{result.url}")
return "\n".join(lines)
def is_hermes_agent_creation_question(message: str) -> bool:
lowered = message.strip().lower()
return "hermes agent" in lowered and any(word in lowered for word in {"kapan", "dibuat", "di buat", "rilis"})
def hermes_agent_creation_reply() -> str:
return (
"Saya belum punya data resmi yang pasti tentang tanggal pertama Hermes Agent dibuat atau dirilis. "
"Jadi saya tidak akan mengarang tanggal. Kalau kamu ingin data yang benar-benar diverifikasi, "
"kirim: carikan informasi resmi tentang kapan Hermes Agent dibuat."
)