745 lines
31 KiB
Python
745 lines
31 KiB
Python
import json
|
|
import socket
|
|
import threading
|
|
from lib import gadget, moop
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
|
|
class StreamCancelled(Exception):
|
|
"""User meminta pembatalan saat stream berjalan.
|
|
|
|
`partial` menyimpan teks yang sudah ter-stream supaya tidak hilang.
|
|
Sengaja bukan turunan _LLMFail: rantai fallback/auto-switch tidak boleh
|
|
memperlakukan cancel sebagai kegagalan provider lalu mencoba model
|
|
berikutnya (jumlah percobaan tidak terbatas)."""
|
|
|
|
def __init__(self, partial=""):
|
|
super().__init__("Stream cancelled by user")
|
|
self.partial = partial
|
|
|
|
|
|
class _CancelState:
|
|
"""Penghubung thread UI (yang membatalkan) dan thread agen (pembaca SSE).
|
|
|
|
Masalah yang dipecahkan: pembacaan stream memblokir di dalam recv() selama
|
|
provider diam — model masih 'thinking' sebelum token pertama, header belum
|
|
dikirim, atau koneksi menggantung. Flag cancel saja tidak cukup karena ia
|
|
hanya diperiksa ketika satu baris penuh tiba, jadi Cancel/Stop baru
|
|
bereaksi setelah provider mengirim sesuatu (bisa puluhan detik, atau tidak
|
|
pernah).
|
|
|
|
Solusi: objek socket koneksi aktif dicatat begitu TCP/TLS terbentuk; saat
|
|
cancel diminta, shutdown(SHUT_RDWR) pada objek itu membangunkan recv()
|
|
seketika di thread mana pun. Yang dipegang adalah objek socket-nya (bukan
|
|
fd mentah), sehingga balapan antara cancel dan penutupan koneksi tidak
|
|
pernah mengenai fd yang sudah dipakai ulang oleh koneksi lain.
|
|
|
|
Beberapa stream bisa berjalan bersamaan pada satu klien (Telegram/XMPP
|
|
memproses tiap pesan di thread sendiri dan berbagi satu LLMClient), jadi
|
|
semua socket aktif dilacak dan dibangunkan sekaligus — persis seperti flag
|
|
global lama yang dilihat oleh setiap loop."""
|
|
|
|
def __init__(self):
|
|
self._lock = threading.Lock()
|
|
self._socks = []
|
|
self.requested = False
|
|
|
|
def set_socket(self, sock):
|
|
"""Daftarkan socket stream aktif. Jika cancel sudah diminta sebelum
|
|
stream sempat terbentuk, bangunkan sekarang juga."""
|
|
wake = None
|
|
with self._lock:
|
|
if self.requested:
|
|
wake = [sock]
|
|
else:
|
|
# Bersihkan entri yang socket-nya sudah ditutup (mis. sisa
|
|
# stream yang selesai tanpa sempat release) supaya daftar tidak
|
|
# tumbuh tanpa batas.
|
|
self._socks = [s for s in self._socks if not self._dead(s)]
|
|
if sock not in self._socks:
|
|
self._socks.append(sock)
|
|
if wake:
|
|
self._shutdown_all(wake)
|
|
|
|
def release(self, sock):
|
|
"""Panggil ketika pembacaan selesai: setelah ini cancel tidak pernah
|
|
menyentuh socket ini lagi. Hanya melepas satu socket — stream thread
|
|
lain pada klien yang sama tidak boleh ikut terputus pendaftarannya."""
|
|
if sock is None:
|
|
return
|
|
with self._lock:
|
|
if sock in self._socks:
|
|
self._socks.remove(sock)
|
|
|
|
def request(self):
|
|
with self._lock:
|
|
self.requested = True
|
|
socks = self._socks
|
|
self._socks = []
|
|
self._shutdown_all(socks)
|
|
|
|
def reset(self):
|
|
# Daftar socket TIDAK dikosongkan: pada klien yang dipakai bersama
|
|
# (Telegram/XMPP) bisa ada stream thread lain yang masih berjalan dan
|
|
# harus tetap bisa dibatalkan. Socket mati tersapu sendiri oleh
|
|
# set_socket/release.
|
|
with self._lock:
|
|
self.requested = False
|
|
|
|
def _shutdown_all(self, socks):
|
|
for s in socks:
|
|
self._shutdown(s)
|
|
|
|
@staticmethod
|
|
def _dead(sock):
|
|
"""True jika socket sudah ditutup (fileno() tidak bisa dipakai lagi)."""
|
|
try:
|
|
sock.fileno()
|
|
except (OSError, ValueError):
|
|
return True
|
|
return False
|
|
|
|
@staticmethod
|
|
def _shutdown(sock):
|
|
"""Putus koneksi sekarang juga. Aman dipanggil dua kali atau pada
|
|
socket yang sudah ditutup: efek yang dibutuhkan hanya membangunkan
|
|
recv() yang terblokir — sekaligus menghentikan generation di sisi
|
|
provider, jadi token tidak ditagih untuk output yang dibuang."""
|
|
try:
|
|
sock.shutdown(socket.SHUT_RDWR)
|
|
except (OSError, ValueError):
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class _CapturingHandler:
|
|
"""Campuran handler urllib: mendaftarkan socket koneksi ke _CancelState
|
|
tepat setelah connect(), sehingga cancel bisa membangunkan pembacaan
|
|
header maupun body yang terblokir."""
|
|
|
|
def __init__(self, state, base_class):
|
|
super().__init__()
|
|
self._state = state
|
|
self._base = base_class
|
|
|
|
def do_open(self, http_class, req, **http_conn_args):
|
|
state = self._state
|
|
captured = []
|
|
|
|
class _Conn(http_class):
|
|
def connect(self):
|
|
http_class.connect(self)
|
|
sock = getattr(self, "sock", None)
|
|
if sock is not None:
|
|
captured.append(sock)
|
|
state.set_socket(sock)
|
|
|
|
try:
|
|
return self._base.do_open(self, _Conn, req, **http_conn_args)
|
|
except BaseException:
|
|
# do_open menutup socket-nya sendiri saat gagal: lepas pendaftaran
|
|
# supaya thread cancel tidak menyentuhnya lagi.
|
|
for sock in captured:
|
|
state.release(sock)
|
|
raise
|
|
|
|
|
|
class _HTTPHandler(_CapturingHandler, urllib.request.HTTPHandler):
|
|
def __init__(self, state):
|
|
_CapturingHandler.__init__(self, state, urllib.request.HTTPHandler)
|
|
|
|
|
|
class _HTTPSHandler(_CapturingHandler, urllib.request.HTTPSHandler):
|
|
def __init__(self, state):
|
|
_CapturingHandler.__init__(self, state, urllib.request.HTTPSHandler)
|
|
|
|
|
|
class _LLMFail(Exception):
|
|
"""
|
|
- 'auth' : 401/403 → rotasi key (coba key berikutnya di provider yang sama)
|
|
- 'unreachable' : koneksi gagal / timeout / 5xx / 429 / 404 → pindah model/provider
|
|
"""
|
|
def __init__(self, kind, message):
|
|
super().__init__(message)
|
|
self.kind = kind
|
|
self.message = message
|
|
|
|
class LLMClient:
|
|
|
|
def __init__(self, base_url, model, api_key, timeout=600):
|
|
self.base_url = (base_url or "").rstrip('/')
|
|
self.model = model
|
|
self.api_key = api_key
|
|
self.timeout = timeout
|
|
|
|
# State cancel + opener yang menangkap socket koneksi (_CancelState).
|
|
self._cancel = _CancelState()
|
|
self._opener = urllib.request.build_opener(
|
|
_HTTPHandler(self._cancel), _HTTPSHandler(self._cancel))
|
|
|
|
# Chaining kandidat set model per tipe ('llm', 'imagevision', ...)
|
|
self._chains: dict[str, list[dict]] = {}
|
|
self._chain: list[dict] | None = None
|
|
self._chain_key: str | None = None
|
|
self._chain_index = 0
|
|
|
|
# -------------------------------------------------------- cancellation ---
|
|
|
|
@property
|
|
def cancel_requested(self) -> bool:
|
|
return self._cancel.requested
|
|
|
|
@cancel_requested.setter
|
|
def cancel_requested(self, value):
|
|
# Kompatibel dengan pemanggil lama (`llm.cancel_requested = True/False`).
|
|
if value:
|
|
self.cancel()
|
|
else:
|
|
self.begin_turn()
|
|
|
|
def cancel(self):
|
|
"""Minta inference berhenti. Aman dipanggil dari thread UI mana pun,
|
|
kapan pun — termasuk saat stream belum terbentuk atau tool sedang
|
|
berjalan. Membangunkan pembacaan socket yang terblokir seketika."""
|
|
self._cancel.request()
|
|
|
|
def begin_turn(self):
|
|
"""Bersihkan state cancel pada awal turn baru (dipanggil interface saat
|
|
user mengirim pesan). Cancel turn sebelumnya tidak boleh merembes ke
|
|
turn berikutnya."""
|
|
self._cancel.reset()
|
|
|
|
def _cancelled_message(self, partial=""):
|
|
"""Respons untuk turn yang dibatalkan user.
|
|
|
|
Teks parsial dikembalikan apa adanya di `content` dan ditandai lewat
|
|
atribut `cancelled` — penanda visual ("dibatalkan") adalah urusan
|
|
interface, bukan bagian dari isi pesan. Rantai model tidak dianggap
|
|
gagal: cancel adalah perintah user, bukan masalah provider."""
|
|
result = self.Message({'content': partial or "", 'tool_calls': None})
|
|
result.cancelled = True
|
|
return result
|
|
|
|
class Message:
|
|
def __init__(self, msg, model_info=None):
|
|
raw_content = msg.get('content', '') # Ambil konten mentah
|
|
self.content = gadget.strip_thinking(raw_content) if isinstance(raw_content, str) else raw_content # Auto-strip <thinking> dari content
|
|
self.tool_calls = msg.get('tool_calls', None) # Ambil tool calls
|
|
self.warning = None
|
|
# True jika pesan ini adalah hasil turn yang dihentikan user.
|
|
self.cancelled = False
|
|
# Pemakaian token dari provider (prompt/completion/total) jika ada.
|
|
self.usage = None
|
|
# Model yang benar-benar dipakai menghasilkan pesan ini. Bisa berbeda
|
|
# dengan model awal permintaan karena auto-switch/fallback.
|
|
self.model_info = model_info or {}
|
|
|
|
# ------------------------------------------------------------ config ---
|
|
|
|
def set_chains(self, chains):
|
|
"""Set chain kandidat per tipe. Memperbarui base_url/model/api_key dari
|
|
kandidat pertama chain 'llm' (atau 'imagevision' sebagai fallback)."""
|
|
self._chains = chains or {}
|
|
self._chain = None
|
|
self._chain_key = None
|
|
self._chain_index = 0
|
|
llm = self._chains.get("llm") or []
|
|
vision = self._chains.get("imagevision") or []
|
|
if llm:
|
|
self._apply_candidate(llm[0])
|
|
elif vision:
|
|
self._apply_candidate(vision[0])
|
|
else:
|
|
self.base_url = ""
|
|
self.model = ""
|
|
self.api_key = ""
|
|
|
|
def set_type_chain(self, type_name, chain):
|
|
"""Perbarui satu tipe chain tanpa menimpa tipe lain.
|
|
|
|
Setelah update, kandidat pertama chain 'llm' kembali diterapkan
|
|
ke base_url/model/api_key aktif (jika ada)."""
|
|
chains = dict(self._chains or {})
|
|
chains[type_name] = chain or []
|
|
self._chains = chains
|
|
self._chain = None
|
|
self._chain_key = None
|
|
self._chain_index = 0
|
|
llm = self._chains.get("llm") or []
|
|
vision = self._chains.get("imagevision") or []
|
|
if llm:
|
|
self._apply_candidate(llm[0])
|
|
elif vision:
|
|
self._apply_candidate(vision[0])
|
|
else:
|
|
self.base_url = ""
|
|
self.model = ""
|
|
self.api_key = ""
|
|
|
|
def has_chain(self, type_name="llm"):
|
|
return bool(self._chains.get(type_name))
|
|
|
|
def _apply_candidate(self, cand):
|
|
self.base_url = (cand.get("base_url") or "").rstrip('/')
|
|
self.model = cand.get("model", "")
|
|
self.api_key = cand.get("api_key", "")
|
|
|
|
def _model_info(self, cand=None):
|
|
"""Info model yang dipakai untuk satu permintaan.
|
|
|
|
Dari kandidat chain jika ada (provider sudah dibawa moop), jika tidak
|
|
(klien non-chain) pakai state aktif dan cari nama provider-nya."""
|
|
if cand:
|
|
return {
|
|
"provider": cand.get("provider"),
|
|
"base_url": cand.get("base_url", ""),
|
|
"model": cand.get("model", ""),
|
|
}
|
|
return {
|
|
"provider": moop.resolve_provider(None, self.base_url, self.model),
|
|
"base_url": self.base_url,
|
|
"model": self.model,
|
|
}
|
|
|
|
# ----------------------------------------------------- chain selection ---
|
|
|
|
def _pick_chain_type(self, messages):
|
|
"""Kalau ada konten gambar (image_url) di tur saat ini, pakai chain
|
|
imagevision. Hanya pesan dari pesan user terakhir ke depan yang ditinjau,
|
|
supaya gambar lama di riwayat tidak memaksa vision untuk teks berikutnya."""
|
|
msgs = messages or []
|
|
last_user = 0
|
|
for i, m in enumerate(msgs):
|
|
if m.get("role") == "user":
|
|
last_user = i
|
|
for m in msgs[last_user:]:
|
|
content = m.get("content")
|
|
if isinstance(content, list):
|
|
for block in content:
|
|
if isinstance(block, dict) and block.get("type") == "image_url":
|
|
return "imagevision"
|
|
return "llm"
|
|
|
|
def chat(self, messages, tools=None, on_stream_chunk=None, disable_reasoning=False, tool_reminder=None):
|
|
|
|
if self._cancel.requested:
|
|
# Cancel diminta sebelum permintaan sempat berangkat (mis. saat
|
|
# tool masih berjalan di langkah sebelumnya). Jangan mulai request
|
|
# baru — kembalikan turn yang ditandai sebagai dibatalkan.
|
|
return self._cancelled_message("")
|
|
|
|
if not self._chains: # kalau bukan chain
|
|
try:
|
|
result = self._chat_once(
|
|
self.base_url, self.model, self.api_key,
|
|
messages,
|
|
tools=tools,
|
|
on_stream_chunk=on_stream_chunk,
|
|
disable_reasoning=disable_reasoning,
|
|
tool_reminder=tool_reminder,
|
|
)
|
|
except StreamCancelled as e:
|
|
return self._cancelled_message(e.partial)
|
|
result.model_info = self._model_info()
|
|
return result
|
|
|
|
chain_type = self._pick_chain_type(messages)
|
|
if chain_type != self._chain_key:
|
|
self._chain_key = chain_type
|
|
self._chain_index = 0
|
|
chain = self._chains.get(chain_type) or []
|
|
self._chain = chain
|
|
|
|
if not chain:
|
|
# Vision diminta tapi tidak ada chain imagevision → pakai llm.
|
|
if chain_type == "imagevision" and self._chains.get("llm"):
|
|
chain = self._chains["llm"]
|
|
self._chain = chain
|
|
self._chain_key = "llm"
|
|
else:
|
|
return self.Message({
|
|
'content': "Error: Tidak ada model yang dikonfigurasi untuk permintaan ini.",
|
|
'tool_calls': None,
|
|
})
|
|
|
|
return self._chat_chain(
|
|
messages, tools=tools, on_stream_chunk=on_stream_chunk,
|
|
disable_reasoning=disable_reasoning, tool_reminder=tool_reminder,
|
|
)
|
|
|
|
# ---------------------------------------------------------- chain loop ---
|
|
|
|
@staticmethod
|
|
def _same_target(a, b):
|
|
return a.get("api_id") == b.get("api_id") and a.get("model_id") == b.get("model_id")
|
|
|
|
@staticmethod
|
|
def _next_target(chain, idx):
|
|
"""Lompat ke kandidat model/provider berikutnya (skip sisa key)."""
|
|
api = chain[idx].get("api_id")
|
|
model = chain[idx].get("model_id")
|
|
j = idx + 1
|
|
while j < len(chain) and chain[j].get("api_id") == api and chain[j].get("model_id") == model:
|
|
j += 1
|
|
return j
|
|
|
|
def _chat_chain(self, messages, tools=None, on_stream_chunk=None,
|
|
disable_reasoning=False, tool_reminder=None):
|
|
chain = self._chain or []
|
|
if not chain:
|
|
return self.Message({'content': "Error: Tidak ada model yang dikonfigurasi.", 'tool_calls': None})
|
|
|
|
idx = min(max(self._chain_index, 0), len(chain) - 1)
|
|
last = None
|
|
|
|
while idx < len(chain):
|
|
cand = chain[idx]
|
|
self._apply_candidate(cand)
|
|
try:
|
|
result = self._chat_once(
|
|
cand["base_url"], cand["model"], cand["api_key"], messages,
|
|
tools=tools, on_stream_chunk=on_stream_chunk,
|
|
disable_reasoning=disable_reasoning, tool_reminder=tool_reminder,
|
|
)
|
|
self._chain_index = idx
|
|
# Catat model yang benar-benar menjawab (bukan model awal
|
|
# permintaan) supaya pemanggil bisa menampilkan data yang jujur.
|
|
result.model_info = self._model_info(cand)
|
|
return result
|
|
except StreamCancelled as e:
|
|
# Cancel user: hentikan rantai total — bukan kegagalan provider,
|
|
# jadi jangan lompat ke kandidat model berikutnya.
|
|
self._chain_index = idx
|
|
result = self._cancelled_message(e.partial)
|
|
result.model_info = self._model_info(cand)
|
|
return result
|
|
except _LLMFail as e:
|
|
last = e
|
|
if e.kind == "auth" and idx + 1 < len(chain) and self._same_target(chain[idx + 1], cand):
|
|
idx += 1 # rotasi key
|
|
else:
|
|
idx = self._next_target(chain, idx) # pindah model/provider
|
|
|
|
# Image vision gagal total → fallback ke model llm umum.
|
|
if self._chain_key == "imagevision":
|
|
fb = self._chains.get("llm") or []
|
|
if fb:
|
|
self._chain = fb
|
|
self._chain_key = "llm"
|
|
self._chain_index = 0
|
|
result = self._chat_chain(
|
|
messages, tools=tools, on_stream_chunk=on_stream_chunk,
|
|
disable_reasoning=disable_reasoning, tool_reminder=tool_reminder,
|
|
)
|
|
result.warning = (f"{result.warning} " if result.warning else "") + \
|
|
"Image vision model gagal, fallback ke LLM umum."
|
|
return result
|
|
|
|
msg = "Error: Semua opsi model gagal."
|
|
if last:
|
|
msg += f"\n{last.message}"
|
|
return self.Message({'content': msg, 'tool_calls': None})
|
|
|
|
# ------------------------------------------------------- single attempt ---
|
|
|
|
def _chat_once(self, base_url, model, api_key, messages, tools=None,
|
|
on_stream_chunk=None, disable_reasoning=False, tool_reminder=None,
|
|
include_usage=True):
|
|
url = f"{base_url.rstrip('/')}/chat/completions"
|
|
# Sanitasi: buang metadata display (model_info/usage) dari pesan sebelum
|
|
# jadi payload — beberapa provider memvalidasi field pesan dengan ketat.
|
|
msgs_clean = [
|
|
{k: v for k, v in m.items() if k not in ("model_info", "usage")}
|
|
if isinstance(m, dict) else m
|
|
for m in (messages or [])
|
|
]
|
|
payload = {
|
|
"model": model,
|
|
"messages": msgs_clean,
|
|
"stream": True # Enable streaming
|
|
}
|
|
if include_usage:
|
|
# Minta provider menyertakan pemakaian token (input/output) pada
|
|
# chunk terakhir stream. Provider yang tidak mendukungnya akan
|
|
# di-retry tanpa parameter ini (lihat penanganan HTTP 400).
|
|
payload["stream_options"] = {"include_usage": True}
|
|
if tools:
|
|
payload["tools"] = tools
|
|
payload["tool_choice"] = "auto"
|
|
if tool_reminder:
|
|
# Inject tool reminder sebagai system message sebelum user message terakhir
|
|
for i in range(len(msgs_clean) - 1, -1, -1):
|
|
if msgs_clean[i].get("role") == "user":
|
|
payload["messages"] = msgs_clean[:i] + [
|
|
{"role": "system", "content": tool_reminder}
|
|
] + msgs_clean[i:]
|
|
break
|
|
|
|
# Hanya kirim parameter reasoning jika diminta eksplisit
|
|
# Beberapa model/provider justru error jika parameter ini ada tapi tidak didukung
|
|
if disable_reasoning:
|
|
payload["reasoning"] = {"enabled": False}
|
|
|
|
data = json.dumps(payload).encode('utf-8')
|
|
req = urllib.request.Request(url, data=data, method='POST')
|
|
req.add_header('Content-Type', 'application/json')
|
|
if api_key:
|
|
req.add_header('Authorization', f'Bearer {api_key}')
|
|
|
|
# Variabel untuk mengumpulkan hasil
|
|
full_content = ""
|
|
full_tool_calls = []
|
|
reasoning_content = ""
|
|
full_multimodal_content = []
|
|
stream_started = False
|
|
usage = None # prompt/completion/total tokens dari chunk terakhir
|
|
|
|
my_sock = None
|
|
try:
|
|
with self._opener.open(req, timeout=self.timeout) as resp:
|
|
# Cadangan pendaftaran socket (jika handler capture tidak
|
|
# sempat mengisi, mis. perubahan internal urllib): cukup untuk
|
|
# membangunkan pembacaan body.
|
|
my_sock = self._register_socket(resp)
|
|
try:
|
|
# Streaming: baca line by line
|
|
for line in resp:
|
|
if self._cancel.requested:
|
|
raise StreamCancelled(full_content)
|
|
|
|
line = line.decode('utf-8').strip()
|
|
if not line or not line.startswith('data: '):
|
|
continue
|
|
|
|
data_str = line[6:] # Hapus "data: " prefix
|
|
if data_str == '[DONE]':
|
|
break
|
|
|
|
try:
|
|
chunk = json.loads(data_str)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
# Pemakaian token datang di chunk penutup (choices kosong
|
|
# pada banyak provider) — ambil selama masih None / terbaru.
|
|
if chunk.get('usage'):
|
|
usage = chunk['usage']
|
|
|
|
# Parse delta dari chunk
|
|
choices = chunk.get('choices') or []
|
|
delta = choices[0].get('delta', {}) if choices else {}
|
|
finish_reason = choices[0].get('finish_reason', None) if choices else None
|
|
|
|
# Stream reasoning content jika ada
|
|
if 'reasoning_content' in delta:
|
|
reasoning_content += delta['reasoning_content']
|
|
|
|
# Stream tool_calls jika ada
|
|
if 'tool_calls' in delta:
|
|
tool_calls = delta['tool_calls']
|
|
for tc in tool_calls:
|
|
idx = tc.get('index', 0)
|
|
# Pastikan list cukup panjang
|
|
while len(full_tool_calls) <= idx:
|
|
full_tool_calls.append({
|
|
"id": "",
|
|
"type": "function",
|
|
"function": {"name": "", "arguments": ""}
|
|
})
|
|
|
|
# Update ID
|
|
if 'id' in tc and tc['id']:
|
|
full_tool_calls[idx]['id'] = tc['id']
|
|
|
|
# Update function name
|
|
if 'function' in tc and 'name' in tc['function']:
|
|
full_tool_calls[idx]['function']['name'] += tc['function']['name']
|
|
|
|
# Update arguments
|
|
if 'function' in tc and 'arguments' in tc['function']:
|
|
full_tool_calls[idx]['function']['arguments'] += tc['function']['arguments']
|
|
|
|
# Stream content (text response atau multimodal blocks)
|
|
if 'content' in delta:
|
|
chunk = delta['content']
|
|
if isinstance(chunk, str):
|
|
full_content += chunk or ""
|
|
if chunk:
|
|
stream_started = True
|
|
if on_stream_chunk and chunk:
|
|
on_stream_chunk(chunk)
|
|
elif isinstance(chunk, list):
|
|
full_multimodal_content.extend(chunk)
|
|
for block in chunk:
|
|
if isinstance(block, dict) and block.get('type') == 'text':
|
|
t = block.get('text', '')
|
|
if t:
|
|
full_content += t
|
|
stream_started = True
|
|
if on_stream_chunk:
|
|
on_stream_chunk(t)
|
|
finally:
|
|
# Lepas pendaftaran begitu pembacaan selesai: socket akan
|
|
# ditutup oleh with-block, dan thread cancel tidak boleh
|
|
# menyentuhnya lagi. Hanya socket milik panggilan ini —
|
|
# stream thread lain pada klien yang sama tetap terdaftar.
|
|
self._cancel.release(my_sock)
|
|
my_sock = None
|
|
|
|
# shutdown() dari thread cancel bisa membuat pembacaan berakhir rapi
|
|
# seperti EOF — pastikan itu tidak disalahartikan sebagai jawaban
|
|
# lengkap.
|
|
if self._cancel.requested:
|
|
raise StreamCancelled(full_content)
|
|
|
|
# Build final response
|
|
if full_multimodal_content:
|
|
message = {'content': full_multimodal_content}
|
|
else:
|
|
message = {'content': full_content}
|
|
|
|
if full_tool_calls:
|
|
# Filter tool_calls yang valid (ada name dan arguments)
|
|
valid_tool_calls = []
|
|
for tc in full_tool_calls:
|
|
name = tc.get('function', {}).get('name')
|
|
args_str = tc.get('function', {}).get('arguments')
|
|
tc_id = tc.get('id')
|
|
|
|
# Pastikan name dan arguments ada
|
|
if name and args_str and args_str.strip():
|
|
# Generate ID jika kosong
|
|
if not tc_id:
|
|
tc_id = f"call_{len(valid_tool_calls)}"
|
|
tc['id'] = tc_id
|
|
|
|
try:
|
|
# Validate dan re-encode JSON untuk format yang konsisten
|
|
parsed_args = json.loads(args_str)
|
|
tc['function']['arguments'] = json.dumps(parsed_args, ensure_ascii=False)
|
|
valid_tool_calls.append(tc)
|
|
except json.JSONDecodeError:
|
|
# Invalid JSON, coba raw string tapi hanya jika tidak kosong
|
|
tc['function']['arguments'] = args_str
|
|
valid_tool_calls.append(tc)
|
|
|
|
if valid_tool_calls:
|
|
message['tool_calls'] = valid_tool_calls
|
|
|
|
response = {'choices': [{'message': message}]}
|
|
except StreamCancelled:
|
|
raise
|
|
except urllib.error.HTTPError as e:
|
|
if self._cancel.requested:
|
|
raise StreamCancelled(full_content)
|
|
body_text = ""
|
|
try:
|
|
body_text = e.read().decode('utf-8', errors='replace')
|
|
except Exception:
|
|
pass
|
|
if tools and e.code == 404:
|
|
try:
|
|
body = json.loads(body_text) if body_text else {}
|
|
if 'tool use' in body.get('error', {}).get('message', '').lower():
|
|
result = self._chat_once(
|
|
base_url, model, api_key, messages, tools=None,
|
|
on_stream_chunk=on_stream_chunk,
|
|
disable_reasoning=disable_reasoning,
|
|
)
|
|
result.warning = "Tool calling not supported by this model. Running in chat-only mode."
|
|
return result
|
|
except Exception:
|
|
pass
|
|
if include_usage and e.code in (400, 422) and "stream_options" in (body_text or "").lower():
|
|
# Provider menolak parameter stream_options — ulangi permintaan
|
|
# tanpanya; token usage tidak tersedia tapi respons tetap jalan.
|
|
return self._chat_once(
|
|
base_url, model, api_key, messages, tools=tools,
|
|
on_stream_chunk=on_stream_chunk,
|
|
disable_reasoning=disable_reasoning,
|
|
tool_reminder=tool_reminder,
|
|
include_usage=False,
|
|
)
|
|
detail = f" - {body_text[:500]}" if body_text else ""
|
|
if e.code in (401, 403):
|
|
raise _LLMFail("auth", f"HTTP {e.code} {e.reason}{detail}")
|
|
raise _LLMFail("unreachable", f"HTTP {e.code} {e.reason}{detail}")
|
|
except urllib.error.URLError as e:
|
|
if self._cancel.requested:
|
|
raise StreamCancelled(full_content)
|
|
if stream_started:
|
|
return self.Message({'content': f"Error: {e.reason}", 'tool_calls': None})
|
|
raise _LLMFail("unreachable", f"Connection error: {e.reason}")
|
|
except (TimeoutError, socket.timeout) as e:
|
|
if self._cancel.requested:
|
|
raise StreamCancelled(full_content)
|
|
if stream_started:
|
|
return self.Message({'content': f"Error: Timeout", 'tool_calls': None})
|
|
raise _LLMFail("unreachable", f"Timeout")
|
|
except Exception as e:
|
|
if self._cancel.requested:
|
|
raise StreamCancelled(full_content)
|
|
if stream_started:
|
|
return self.Message({'content': f"Error: {str(e)}", 'tool_calls': None})
|
|
raise _LLMFail("unreachable", f"Error: {str(e)}")
|
|
|
|
if 'choices' not in response:
|
|
raw_preview = json.dumps(response)[:500]
|
|
raise _LLMFail("unreachable", (
|
|
f"Unexpected response — 'choices' key missing.\n"
|
|
f" URL : {url}\n"
|
|
f" Model : {model}\n"
|
|
f" Response: {raw_preview}"
|
|
))
|
|
if not response['choices']:
|
|
raw_preview = json.dumps(response)[:500]
|
|
raise _LLMFail("unreachable", (
|
|
f"'choices' is empty in the response.\n"
|
|
f" URL : {url}\n"
|
|
f" Model : {model}\n"
|
|
f" Response: {raw_preview}"
|
|
))
|
|
if 'message' not in response['choices'][0]:
|
|
raw_preview = json.dumps(response['choices'][0])[:500]
|
|
raise _LLMFail("unreachable", (
|
|
f"'message' key missing in first choice.\n"
|
|
f" URL : {url}\n"
|
|
f" Model : {model}\n"
|
|
f" Choice : {raw_preview}"
|
|
))
|
|
|
|
message = response['choices'][0]['message']
|
|
|
|
# Handle reasoning_content field dari OpenRouter/models yang support thinking
|
|
# Pindahkan ke content jangan sampai keluar
|
|
message.pop('reasoning_content', None)
|
|
message.pop('reasoning', None)
|
|
|
|
result = self.Message(message)
|
|
result.usage = usage
|
|
return result
|
|
|
|
@staticmethod
|
|
def _socket_of(resp):
|
|
"""Objek socket di balik HTTPResponse; None jika struktur urllib
|
|
berbeda dari yang diharapkan (maka fallback wake tidak tersedia)."""
|
|
try:
|
|
fp = resp.fp
|
|
raw = getattr(fp, "raw", None)
|
|
sock = getattr(raw, "_sock", None)
|
|
return sock if isinstance(sock, socket.socket) else None
|
|
except (AttributeError, TypeError, ValueError):
|
|
return None
|
|
|
|
def _register_socket(self, resp):
|
|
"""Daftarkan socket respons ke state cancel; kembalikan socket-nya
|
|
supaya pemanggil bisa melepasnya sendiri saat selesai (tidak menyentuh
|
|
stream thread lain). Jika cancel sudah diminta, set_socket langsung
|
|
memutusnya sehingga pembacaan pertama tidak sempat terblokir."""
|
|
sock = self._socket_of(resp)
|
|
if sock is not None:
|
|
self._cancel.set_socket(sock)
|
|
return sock
|