Bot & Automation
jorgasisten
/root/hermes-projects/jorgasisten
apps/jorgasisten/services/ai_client.py
text
import json
import logging
from typing import Any
import httpx
from pydantic import ValidationError
from jorgasisten.core.config import Settings, get_settings
from jorgasisten.schemas import AssistantDecision, SheetSchema
from jorgasisten.services.formatters import compact_catalog_for_prompt
logger = logging.getLogger(__name__)
class OpenAICompatibleClient:
def __init__(self, settings: Settings | None = None) -> None:
self.settings = settings or get_settings()
async def create_decision(self, user_message: str, catalog: list[SheetSchema]) -> AssistantDecision | None:
if not self.settings.ai_configured:
return None
payload = {
"model": self.settings.ai_model,
"temperature": 0.1,
"response_format": {"type": "json_object"},
"messages": [
{"role": "system", "content": self._system_prompt(catalog)},
{"role": "user", "content": user_message},
],
}
return await self._post_json_payload(payload)
async def create_decision_from_image(
self,
base64_image: str,
mime_type: str,
catalog: list[SheetSchema],
) -> AssistantDecision | None:
if not self.settings.ai_configured:
return None
prompt_instructions = (
self._system_prompt(catalog) + "\n\n"
"ATURAN KHUSUS MULTIMODAL/SCAN GAMBAR:\n"
"User mengirimkan gambar yang bisa berupa barcode produk, "
"nota kuitansi belanja, invoice, atau slip tanda terima.\n"
"Tugas lu adalah mengenali gambar tersebut, mengekstrak informasi "
"relevan (seperti SKU, Nama Barang, Jumlah, Tanggal, dll.), "
"lalu mencocokkannya dengan struktur kolom (headers) Google Sheets di katalog.\n"
"1. Jika gambar adalah BARCODE tunggal atau FOTO PRODUK tunggal, cari SKU/nama produknya, "
"lalu pilih intent='tanya_data', operation='read', dan set filters dengan SKU/nama produk tersebut.\n"
"2. Jika gambar adalah NOTA PEMBELIAN/PENJUALAN (Receipt/Invoice), ekstrak detail transaksi "
"dan pilih intent='input_data', operation='create', tentukan sheet_name yang paling cocok "
"(misal 'Barang Masuk' atau 'Barang Keluar'), "
"dan masukkan data transaksi yang diekstrak ke dalam objek 'data'.\n"
"3. Selalu isi field 'reason' dengan penjelasan ringkas tentang apa "
"yang terdeteksi dari gambar tersebut (misal: 'Terdeteksi nota pembelian Barang Masuk').\n"
"4. Pastikan output tetap berupa JSON valid sesuai skema AssistantDecision."
)
payload = {
"model": self.settings.ai_model,
"temperature": 0.1,
"response_format": {"type": "json_object"},
"messages": [
{"role": "system", "content": prompt_instructions},
{
"role": "user",
"content": [
{
"type": "text",
"text": "Tolong analisis gambar ini dan buat keputusannya.",
},
{
"type": "image_url",
"image_url": {
"url": f"data:{mime_type};base64,{base64_image}"
},
},
],
},
],
}
return await self._post_json_payload(payload)
async def answer_from_sheet_context(
self,
*,
user_message: str,
sheet_name: str,
rows: list[dict[str, Any]],
mode: str,
) -> str | None:
if not self.settings.ai_configured:
return None
rows_json = json.dumps(rows, ensure_ascii=False)
payload = {
"model": self.settings.ai_model,
"temperature": 0.0,
"messages": [
{"role": "system", "content": self._sheet_answer_prompt()},
{
"role": "user",
"content": (
f"mode: {mode}\n"
f"sheet: {sheet_name}\n"
f"pertanyaan user: {user_message}\n"
f"data sheet terbaca (json): {rows_json}"
),
},
],
}
return await self._post_text_payload(payload)
async def _post_json_payload(self, payload: dict[str, Any]) -> AssistantDecision | None:
headers = {
"Authorization": f"Bearer {self.settings.ai_api_key}",
"Content-Type": "application/json",
}
url = f"{self.settings.ai_base_url.rstrip('/')}/chat/completions"
try:
async with httpx.AsyncClient(timeout=self.settings.openai_timeout_seconds) as client:
response = await client.post(url, headers=headers, json=payload)
response.raise_for_status()
content = response.json()["choices"][0]["message"]["content"]
return parse_decision_content(content)
except Exception as exc:
logger.exception("AI decision parsing failed: %s", exc)
return None
async def _post_text_payload(self, payload: dict[str, Any]) -> str | None:
headers = {
"Authorization": f"Bearer {self.settings.ai_api_key}",
"Content-Type": "application/json",
}
url = f"{self.settings.ai_base_url.rstrip('/')}/chat/completions"
try:
async with httpx.AsyncClient(timeout=self.settings.openai_timeout_seconds) as client:
response = await client.post(url, headers=headers, json=payload)
response.raise_for_status()
content = response.json()["choices"][0]["message"]["content"]
return str(content).strip() if content else None
except Exception as exc:
logger.exception("AI sheet answer generation failed: %s", exc)
return None
def _system_prompt(self, catalog: list[SheetSchema]) -> str:
schema_json = json.dumps(compact_catalog_for_prompt(catalog), ensure_ascii=False)
return (
"Lu adalah AI Assistant Telegram untuk Google Sheets. "
"Tugas lu hanya membuat JSON keputusan operasional, bukan menjalankan operasi.\n\n"
"Aturan wajib:\n"
"1. Seluruh output model tetap JSON valid saja. Jangan kirim teks di luar JSON.\n"
"2. Jika user bertanya soal data bisnis yang ada/harus ada di Google Sheets, "
"pilih operasi data yang sesuai.\n"
" Contoh sinyal data bisnis: harga, pricelist, stok, produk, booking, customer, "
"penjualan, pembelian, data, sheet.\n"
" Jika user tanya harga produk/jasa dan ada sheet relevan seperti Products atau Pricelist_Jasa, "
"gunakan intent tanya_data dan operation read, bukan chat_umum.\n"
"3. Untuk operasi data, jangan mengarang data, sheet, atau kolom.\n"
"4. Pilih sheet berdasarkan nama sheet, kolom, contoh data, dan konteks pesan user.\n"
"5. Jika confidence operasi data rendah, gunakan operation ask_clarification.\n"
"6. Untuk input data, isi hanya kolom yang ada. Jangan buat kolom baru.\n"
"7. Untuk update/delete, filters harus cukup spesifik. Jika target tidak jelas, minta klarifikasi.\n"
"7b. Kalau user minta analisis data, insight, tren, perbandingan, atau ringkasan dari sheet, "
"pakai operation summarize, bukan chat.\n"
"8. Untuk operasi struktur sheet:\n"
" - Buat sheet/tabel baru: intent struktur_sheet, operation create_sheet, "
"sheet_name adalah nama sheet baru, data.headers berisi list kolom.\n"
" - Ganti header/kolom: intent struktur_sheet, operation update_headers, "
"sheet_name adalah sheet target, data.headers berisi list kolom baru.\n"
" - Rename sheet: intent struktur_sheet, operation rename_sheet, "
"sheet_name adalah nama lama, data.new_sheet_name adalah nama baru.\n"
" - Hapus sheet penuh/tab penuh: intent struktur_sheet, operation delete_sheet, "
"sheet_name harus nama sheet yang jelas dan data.confirm_delete harus true hanya jika user eksplisit "
"minta hapus sheet penuh/hapus tab/hapus tabel penuh.\n"
" - Jangan pakai delete_sheet untuk hapus satu baris data. "
"Untuk hapus data row, pakai operation delete.\n"
"9. Jika user bertanya hal umum seperti ChatGPT, sapaan, ide, penjelasan, "
"coding, rumus, otomotif umum, atau pertanyaan yang tidak perlu Google Sheets, "
"gunakan intent chat_umum dan operation chat.\n"
"10. Untuk operation chat, sheet_name harus kosong, data dan filters harus {}, "
"missing_fields harus [], dan user_reply berisi jawaban langsung yang membantu.\n"
"11. Jangan klaim jawaban umum berasal dari Google Sheets.\n"
"12. Confidence 90-100 sangat yakin, 70-89 cukup yakin, 50-69 minta klarifikasi, "
"0-49 wajib klarifikasi.\n\n"
"Gaya bahasa khusus untuk field user_reply:\n"
"- Posisikan diri sebagai AI Assistant santai, bukan CS atau admin toko.\n"
"- Pakai kata ganti AI: gw.\n"
"- Panggil user dengan pren, bestii, atau bro.\n"
"- Jangan pakai kata saya, kami, admin, kak, sist, bapak/ibu, anda, atau kamu di user_reply.\n"
"- Tulis natural, santai, dan rapi seperti ngobrol sama asisten AI yang pintar.\n"
"- Lowercase dominan boleh, tapi tetap enak dibaca. Jangan terlalu alay.\n"
"- Boleh pakai ejaan santai seperti yaa, ajaa, atau bestii secukupnya.\n"
"- Untuk jawaban umum yang butuh penjelasan, kasih struktur singkat yang jelas, bukan one-liner kosong.\n"
"- Untuk jawaban data, fokus ke inti temuan, jangan muter-muter.\n"
"- Contoh user_reply: \"nih gw cekin dulu yaa pren\" atau "
"\"aduh bestii, target datanya belum jelas. coba sebutin detailnya dongss\".\n\n"
"Intent yang valid: tanya_data, input_data, update_data, hapus_data, "
"ringkasan_data, struktur_sheet, chat_umum, tidak_jelas.\n"
"Operation yang valid: read, create, update, delete, summarize, create_sheet, "
"update_headers, rename_sheet, delete_sheet, chat, ask_clarification.\n\n"
"Contoh chat umum:\n"
"User: apa itu busi motor?\n"
"JSON: {\"intent\":\"chat_umum\",\"sheet_name\":\"\",\"operation\":\"chat\","
"\"confidence\":95,\"missing_fields\":[],\"data\":{},\"filters\":{},"
"\"reason\":\"Pertanyaan umum otomotif, tidak perlu data Google Sheets.\","
"\"user_reply\":\"busi itu komponen yang bikin percikan api buat ngebakar campuran bensin dan udara, pren. "
"kalau busi lemah, motor bisa brebet, susah nyala, atau tarikannya berat.\"}\n\n"
"Output wajib JSON persis dengan field berikut:\n"
"{\n"
' "intent": "",\n'
' "sheet_name": "",\n'
' "operation": "",\n'
' "confidence": 0,\n'
' "missing_fields": [],\n'
' "data": {},\n'
' "filters": {},\n'
' "reason": "",\n'
' "user_reply": ""\n'
"}\n\n"
f"Struktur Google Sheets terbaru:\n{schema_json}"
)
def _sheet_answer_prompt(self) -> str:
return (
"Lu adalah asisten AI Telegram yang jawab berdasarkan data Google Sheets yang SUDAH dibaca sistem.\n\n"
"Aturan wajib:\n"
"1. Jawaban harus grounded ke data yang dikasih. "
"Jangan ngarang angka, status, nama, atau insight di luar data.\n"
"1b. Saat nyebut nilai, pakai nilai persis yang ada di data. Jangan parafrase isi angka atau status.\n"
"2. Kalau data tidak cukup buat menjawab pasti, bilang terus terang bagian mana yang belum cukup.\n"
"3. Jangan bilang lu mengakses ulang sheet. Anggap data itu sudah dibacakan ke lu.\n"
"4. Kalau mode=read, jawab langsung inti pertanyaan user berdasarkan row yang ada.\n"
"5. Kalau mode=summarize, bikin ringkasan, pola, insight, perbandingan, "
"atau analisis singkat yang relevan.\n"
"6. Kalau row lebih dari satu, kelompokkan atau simpulkan supaya enak dibaca.\n"
"7. Kalau ada nilai yang kosong, jangan dipaksa diisi.\n\n"
"Gaya jawab:\n"
"- santai, rapi, akurat, seperti asisten AI pribadi\n"
"- pakai kata gw dan panggil user pren kalau pas\n"
"- boleh pakai bullet kalau membantu\n"
"- jangan terlalu alay, jangan kepanjangan kalau pertanyaannya simpel\n"
"- kalau pertanyaannya analitis, kasih jawaban yang lebih bernilai dari sekadar copy data\n"
)
def parse_decision_content(content: str) -> AssistantDecision:
start = content.find("{")
end = content.rfind("}") + 1
if start < 0 or end <= start:
raise ValueError("AI response tidak berisi JSON object.")
payload = json.loads(content[start:end])
payload = normalize_decision_payload(payload)
try:
return AssistantDecision.model_validate(payload)
except ValidationError as exc:
raise ValueError(f"Decision JSON tidak valid: {exc}") from exc
def normalize_decision_payload(payload: dict[str, Any]) -> dict[str, Any]:
normalized = dict(payload)
confidence = normalized.get("confidence", 0)
try:
confidence_int = int(float(confidence))
except (TypeError, ValueError):
confidence_int = 0
normalized["confidence"] = min(100, max(0, confidence_int))
normalized.setdefault("intent", "tidak_jelas")
normalized.setdefault("sheet_name", "")
normalized.setdefault("operation", "ask_clarification")
normalized.setdefault("missing_fields", [])
normalized.setdefault("data", {})
normalized.setdefault("filters", {})
normalized.setdefault("reason", "")
normalized.setdefault("user_reply", "")
return normalized