""" jendela.py — JendelaWacana: akumulasi eksplisit discourse AKSARA. ANTI-TRANSFORMER: Transformer: konteks = KV cache token embeddings (opaque, tidak bisa diaudit) AKSARA: konteks = list BingkaiWacana + entitas tracker + pool inferensi Setiap elemen bisa dibaca, dikoreksi, dan di-patch tanpa retrain. Analogi terbaik: catatan rapat yang diisi saat mendengarkan — bukan rekaman audio yang harus di-replay dari awal setiap kali ingin tahu sesuatu. Operasi utama: tambah(kalimat) → proses kalimat, perbarui konteks, return BingkaiWacana tanya(query) → cari jawaban dari pool inferensi kumulatif ringkas() → proposisi kunci discourse sebagai string siapa(nama) → cari EntitasAktif berdasarkan nama atau variasi """ from __future__ import annotations from typing import Dict, List, Optional, Tuple from aksara.base.state import AksaraState from aksara.primitives.krl.proposition import Proposisi, TipeSlot from aksara.primitives.krl.inference import Inferensi from aksara.wacana.entitas import ( EntitasAktif, BingkaiWacana, PRONOMINA_PERSONA, DEMONSTRATIVA_ANAFORIS, TIPE_PERSONA, ) # Kata tanya dan relasi inferensi yang paling cocok untuk menjawabnya _KATA_TANYA_KE_RELASI: Dict[str, List[str]] = { "status": ["STATUS_MENJADI", "MEMILIKI", "ATRIBUSI"], "mengapa": ["WAJIB", "KAUSAL", "SEBAB"], "siapa": ["AGEN", "PERSONA"], "apa": ["MEMILIKI", "STATUS_MENJADI", "ATRIBUSI"], "bagaimana":["CARA", "PROSES"], "kapan": ["WAKTU", "URUTAN"], "di mana": ["LOKASI", "BERADA_DI"], } class JendelaWacana: """ Jendela konteks discourse AKSARA. Mempertahankan representasi eksplisit discourse lintas kalimat: - Riwayat kalimat beserta AksaraState masing-masing - Tracker entitas aktif dengan resolusi referensi - Pool inferensi kumulatif dari KRL (forward chaining) Tidak ada parameter yang dilatih — semuanya deterministik dan auditable. Contoh penggunaan: jendela = JendelaWacana(framework) jendela.tambah("Hakim menjatuhkan hukuman kepada terdakwa.") jendela.tambah("Dia menangis di ruang sidang.") hasil = jendela.tanya("Apa status terdakwa?") print(hasil["jawaban_proposisi"]) # STATUS_MENJADI(terdakwa, terpidana) """ def __init__(self, framework, batas_jendela: int = 15) -> None: """ Args: framework: AksaraFramework — backbone 6 primitif batas_jendela: jumlah maksimum kalimat yang disimpan dalam jendela """ self.framework = framework self.batas_jendela = batas_jendela self._bingkai: List[BingkaiWacana] = [] self._entitas: Dict[str, EntitasAktif] = {} # nama_kanonis → EntitasAktif self._inferensi: List[Inferensi] = [] # pool kumulatif # ── Operasi utama ────────────────────────────────────────────────────── def tambah(self, kalimat: str) -> BingkaiWacana: """ Tambah kalimat ke jendela wacana. Urutan proses: 1. Jalankan AksaraFramework → AksaraState nyata 2. Ekstrak proposisi dan inferensi dari KRL 3. Resolve referensi pronomina ke entitas yang sudah dikenal 4. Perbarui tracker entitas 5. Akumulasi inferensi ke pool global 6. Simpan bingkai, jaga batas jendela Returns: BingkaiWacana — bingkai yang baru ditambahkan """ indeks = len(self._bingkai) state: AksaraState = self.framework.proses(kalimat) proposisi: Optional[Proposisi] = None inferensi: List[Inferensi] = [] if state.krl_result is not None: proposisi = state.krl_result.proposisi hi = state.krl_result.hasil_inferensi if hi is not None and hasattr(hi, "inferensi"): inferensi = list(hi.inferensi) bingkai = BingkaiWacana( indeks=indeks, kalimat=kalimat, state=state, proposisi=proposisi, inferensi=inferensi, ) # Resolve referensi SEBELUM update entitas agar pronomina masih terdeteksi proposisi = self._resolve_referensi(proposisi, kalimat) bingkai = BingkaiWacana( indeks=indeks, kalimat=kalimat, state=state, proposisi=proposisi, inferensi=inferensi, ) self._update_entitas(bingkai) self._akumulasi_inferensi(bingkai) self._bingkai.append(bingkai) if len(self._bingkai) > self.batas_jendela: self._bingkai = self._bingkai[-self.batas_jendela:] return bingkai def tanya(self, query: str) -> Optional[Dict]: """ Jawab pertanyaan berdasarkan konteks yang sudah diakumulasi. Proses: 1. Parse query via framework → AksaraState + proposisi 2. Ekstrak entitas dan kata tanya dari query 3. Cari inferensi yang relevan di pool 4. Return inferensi terbaik + penjelasannya Returns: Dict dengan kunci: inferensi — Inferensi paling relevan jawaban_proposisi — string representasi inferensi keyakinan — float [0,1] penjelasan — sumber aturan + kalimat asal atau None jika tidak ada jawaban yang ditemukan. """ if not self._inferensi: return None state_q = self.framework.proses(query) entitas_q = self._ekstrak_entitas_query(state_q, query) relasi_cocok = self._relasi_dari_kata_tanya(query) kandidat = self._cari_inferensi_relevan(entitas_q, relasi_cocok, query) if not kandidat: return None terbaik = max(kandidat, key=lambda i: i.keyakinan) return { "inferensi": terbaik, "jawaban_proposisi": f"{terbaik.relasi}({terbaik.subjek}, {terbaik.objek})", "keyakinan": terbaik.keyakinan, "penjelasan": ( f"Berdasarkan aturan '{terbaik.aturan_sumber}', " f"dari kalimat: \"{terbaik.proposisi_asal}\"" ), } def siapa(self, nama: str) -> Optional[EntitasAktif]: """Cari entitas berdasarkan nama atau variasi rujukan.""" for ent in self._entitas.values(): if ent.cocok(nama): return ent return None def apa_yang_terjadi_pada(self, nama_entitas: str) -> List[Inferensi]: """Semua inferensi yang melibatkan entitas tertentu sebagai subjek atau objek.""" nama_lower = nama_entitas.lower() return [ inf for inf in self._inferensi if nama_lower in (inf.subjek.lower(), inf.objek.lower()) ] def ringkas(self) -> str: """ Ringkasan discourse sebagai teks proposisi kunci. Format: [1] AKSI(agen=X, pasien=Y) [2] AKSI(agen=A, pasien=B) Inferensi kumulatif: → RELASI(subjek, objek) [aturan] """ baris = [] for b in self._bingkai: if b.proposisi: baris.append(f"[{b.indeks + 1}] {b.proposisi}") else: baris.append(f"[{b.indeks + 1}] \"{b.kalimat}\" (tanpa proposisi)") if self._inferensi: baris.append("\nInferensi kumulatif:") for inf in self._inferensi[:8]: baris.append( f" → {inf.relasi}({inf.subjek}, {inf.objek})" f" [{inf.aturan_sumber}]" ) if len(self._inferensi) > 8: baris.append(f" ... dan {len(self._inferensi) - 8} inferensi lain") if self._entitas: baris.append("\nEntitas aktif:") for kanonis, ent in self._entitas.items(): variasi = sorted(ent.nama_variasi - {kanonis}) var_str = f" (juga: {', '.join(variasi)})" if variasi else "" baris.append(f" {kanonis} [{ent.tipe}]{var_str}") return "\n".join(baris) if baris else "(wacana kosong)" def reset(self) -> None: """Bersihkan seluruh konteks.""" self._bingkai.clear() self._entitas.clear() self._inferensi.clear() # ── Properti ─────────────────────────────────────────────────────────── @property def n_kalimat(self) -> int: return len(self._bingkai) @property def n_inferensi(self) -> int: return len(self._inferensi) @property def entitas_aktif(self) -> List[EntitasAktif]: return list(self._entitas.values()) @property def inferensi_semua(self) -> List[Inferensi]: return list(self._inferensi) @property def bingkai_terakhir(self) -> Optional[BingkaiWacana]: return self._bingkai[-1] if self._bingkai else None # ── Internals: resolusi referensi ────────────────────────────────────── def _resolve_referensi( self, proposisi: Optional[Proposisi], kalimat: str ) -> Optional[Proposisi]: """ Resolve pronomina dalam proposisi ke entitas kanonik. Jika agen = "dia" dan ada PERSONA aktif di riwayat → ganti dengan nama kanonik. Modifikasi dilakukan pada nilai slot, bukan pada objek Proposisi. """ if proposisi is None or not self._entitas: return proposisi for tipe_slot, slot in proposisi.slot.items(): nilai_lower = slot.nilai.lower() if nilai_lower in PRONOMINA_PERSONA: # Cari entitas persona yang paling baru persona_terakhir = self._cari_persona_terakhir() if persona_terakhir is not None: slot.nilai = persona_terakhir.nama_kanonis persona_terakhir.tambah_variasi(nilai_lower) elif nilai_lower in DEMONSTRATIVA_ANAFORIS: # Cari entitas benda/abstrak yang paling baru benda_terakhir = self._cari_benda_terakhir() if benda_terakhir is not None: slot.nilai = benda_terakhir.nama_kanonis benda_terakhir.tambah_variasi(nilai_lower) return proposisi def _cari_persona_terakhir(self) -> Optional[EntitasAktif]: """Entitas bertipe persona dengan indeks_terakhir tertinggi.""" persona = [e for e in self._entitas.values() if e.adalah_persona()] if not persona: return None return max(persona, key=lambda e: e.indeks_terakhir) def _cari_benda_terakhir(self) -> Optional[EntitasAktif]: """Entitas non-persona (benda/abstrak) yang paling baru.""" benda = [e for e in self._entitas.values() if not e.adalah_persona()] if not benda: return None return max(benda, key=lambda e: e.indeks_terakhir) # ── Internals: update entitas ────────────────────────────────────────── def _update_entitas(self, bingkai: BingkaiWacana) -> None: """Perbarui tracker entitas dari bingkai baru.""" if bingkai.proposisi is None: return prop = bingkai.proposisi krl = bingkai.state.krl_result # Pasang (nilai_slot, tipe_slot) ke tracker for tipe_slot, slot in prop.slot.items(): nama = slot.nilai if not nama or nama.lower() in PRONOMINA_PERSONA | DEMONSTRATIVA_ANAFORIS: continue peran_str = tipe_slot.value entitas_ada = self._cari_entitas(nama) if entitas_ada is not None: entitas_ada.perbarui_kemunculan(bingkai.indeks, peran_str) else: tipe = self._inferensi_tipe(nama, krl) ent = EntitasAktif( nama_kanonis=nama, tipe=tipe, nama_variasi={nama}, indeks_pertama=bingkai.indeks, indeks_terakhir=bingkai.indeks, peran_riwayat=[(bingkai.indeks, peran_str)], ) self._entitas[nama] = ent # Gunakan ikatan referensi dari KRL jika ada if krl is not None and hasattr(krl, "ikatan_referensi"): for ikatan in (krl.ikatan_referensi or []): anafor = getattr(ikatan, "anafor", None) anteseden = getattr(ikatan, "anteseden", None) if anafor and anteseden: ent = self._cari_entitas(anteseden) if ent is not None: ent.tambah_variasi(anafor) def _cari_entitas(self, nama: str) -> Optional[EntitasAktif]: """Cari entitas yang cocok dengan nama (case-insensitive).""" for ent in self._entitas.values(): if ent.cocok(nama): return ent return None def _inferensi_tipe(self, nama: str, krl_result) -> str: """ Inferensikan tipe entitas dari konteks KRL, KB, dan pola nama. Urutan pencarian: 1. KB KATA_KE_TIPE (eksplisit, ~60 kata terpetakan) 2. Proposisi KRL + heuristik agen kapital → PERSONA 3. Fallback heuristik kapital → PERSONA 4. Default → ENTITAS """ # 1. Lookup langsung di Knowledge Base from aksara.primitives.krl.kb import KATA_KE_TIPE tipe_kb = KATA_KE_TIPE.get(nama.lower()) if tipe_kb is not None: return tipe_kb # 2. Konteks KRL: agen dengan nama kapital → PERSONA if krl_result is not None and hasattr(krl_result, "proposisi") and krl_result.proposisi: prop = krl_result.proposisi agen = prop.slot.get(TipeSlot.AGEN) if agen and agen.nilai == nama and nama[0].isupper(): return "PERSONA" # 3. Heuristik: nama proper (kapital awal) → PERSONA if nama and nama[0].isupper(): return "PERSONA" return "ENTITAS" # ── Internals: akumulasi inferensi ───────────────────────────────────── def _akumulasi_inferensi(self, bingkai: BingkaiWacana) -> None: """Tambahkan inferensi baru dari bingkai ke pool — tanpa duplikasi.""" for inf in bingkai.inferensi: duplikat = any( i.relasi == inf.relasi and i.subjek.lower() == inf.subjek.lower() and i.objek.lower() == inf.objek.lower() for i in self._inferensi ) if not duplikat: self._inferensi.append(inf) # ── Internals: pencarian inferensi ───────────────────────────────────── def _cari_inferensi_relevan( self, entitas_query: List[str], relasi_cocok: List[str], query: str, ) -> List[Inferensi]: """ Cari inferensi yang relevan dengan query. Skor relevansi dihitung dari: +0.5 jika subjek atau objek cocok dengan entitas yang ditanya +0.3 jika relasi cocok dengan kata tanya +0.1 jika domain inferensi ada di query """ hasil = [] query_lower = query.lower() for inf in self._inferensi: skor = 0.0 for nama in entitas_query: if nama.lower() in (inf.subjek.lower(), inf.objek.lower()): skor += 0.5 # Cek juga variasi nama ent = self._cari_entitas(nama) if ent: for var in ent.nama_variasi: if var.lower() in (inf.subjek.lower(), inf.objek.lower()): skor += 0.2 break for relasi in relasi_cocok: if relasi.upper() in inf.relasi.upper(): skor += 0.3 if inf.domain != "universal" and inf.domain in query_lower: skor += 0.1 if skor >= 0.3: hasil.append(inf) return sorted(hasil, key=lambda i: i.keyakinan, reverse=True) def _ekstrak_entitas_query( self, state: AksaraState, query: str ) -> List[str]: """Ekstrak entitas yang disebutkan dalam query.""" entitas: List[str] = [] # Dari proposisi KRL query if state.krl_result and state.krl_result.proposisi: prop = state.krl_result.proposisi for slot in prop.slot.values(): if slot.nilai and slot.nilai not in entitas: entitas.append(slot.nilai) # Dari entitas yang sudah dikenal dalam wacana query_lower = query.lower() for kanonis, ent in self._entitas.items(): for var in ent.nama_variasi: if var.lower() in query_lower and kanonis not in entitas: entitas.append(kanonis) break return entitas def _relasi_dari_kata_tanya(self, query: str) -> List[str]: """Mapping kata tanya dalam query ke daftar relasi yang relevan.""" query_lower = query.lower() relasi: List[str] = [] for kata, daftar_relasi in _KATA_TANYA_KE_RELASI.items(): if kata in query_lower: relasi.extend(daftar_relasi) return list(dict.fromkeys(relasi)) # deduplicate, pertahankan urutan