| """TEST AMAÇLI sahte LLM sunucusu — gerçek bir dil modeli DEĞİLDİR. |
| |
| Anthropic (/v1/models, /v1/messages), OpenAI (/v1/models, /v1/chat/completions) ve Ollama (/api/tags, |
| /api/pull, /api/chat, /api/version) uç noktalarının istek/yanıt biçimlerini taklit eder. Böylece model |
| senkronizasyonu, sağlayıcı bağlantıları ve tüm AI özellikleri API anahtarı olmadan uçtan uca sınanabilir. |
| |
| SQL görevlerinde yanıtı projenin kural tabanlı üreticisiyle üretir. Kendini düzeltme döngüsünü göstermek |
| için, soruda gg.aa.yyyy biçimli bir tarih varsa ilk denemede bilerek Türkçe tarih literali yazar. |
| |
| Çalıştırma: python tests/mock_llm_server.py [port] (vars. 8765) |
| Claude : ANTHROPIC_BASE_URL=http://localhost:8765 ANTHROPIC_API_KEY=test |
| OpenAI : Base URL http://localhost:8765/v1 anahtar: test |
| Ollama : OLLAMA_HOST=http://localhost:8765 |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import os |
| import re |
| import sys |
| import time |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parent.parent |
| sys.path.insert(0, str(ROOT)) |
|
|
| from t2s.rulebased import RuleBasedTextToSQL |
| from t2s.schema import connect |
| from t2s.trnum import slugify |
|
|
| LAST = {} |
| DB = os.getenv("MOCK_DB", str(ROOT / "outputs" / "db" / "arayuz.db")) |
| OLLAMA_MODELS = ["qwen2.5-coder:7b", "llama3.1:8b"] |
| |
| |
| EXTRA = os.getenv("MOCK_EXTRA") == "1" |
| EN_DEMO = { |
| "what was izmir's electricity consumption in 2024?": "İzmir'in 2024 elektrik tüketimi ne kadar?", |
| "which province consumed the most electricity in 2024?": "2024'te en çok elektrik tüketen il hangisi?", |
| "how much were personnel expenses in 2024?": "2024 yılında personel giderleri için ne kadar harcama gerçekleşti?", |
| } |
|
|
|
|
| def _sql_reply(messages: list[dict]) -> str: |
| first = messages[0]["content"] |
| m = re.search(r"### Soru\n(.+?)\n", first, re.S) |
| q = (m.group(1) if m else "").strip() |
| q_tr = EN_DEMO.get(q.lower(), q) |
| rb = RuleBasedTextToSQL(connect(DB, readonly=True)) |
| r = rb.generate(q_tr) |
| feedback_turns = sum(1 for x in messages if x["role"] == "user") - 1 |
| if not r.candidates: |
| return json.dumps({"dusunce": "Tabloda karşılık yok.", "cevaplanabilir": False, "sql": "", |
| "gerekce": r.reason}, ensure_ascii=False) |
| sql = r.candidates[0] |
| dm = re.search(r"\b(\d{2})\.(\d{2})\.(\d{4})\b", q_tr) |
| if dm and feedback_turns == 0: |
| iso = f"{dm.group(3)}-{dm.group(2)}-{dm.group(1)}" |
| sql = sql.replace(f"'{iso}'", f"'{dm.group(0)}'") |
| return json.dumps({"dusunce": "Şemadaki ilgili tablo ve sütunlar kullanıldı.", "cevaplanabilir": True, "sql": sql}, |
| ensure_ascii=False) |
|
|
|
|
| def _lang(system: str) -> str: |
| return "en" if "Yanıt dili: English" in system else "tr" |
|
|
|
|
| def reply(system: str, messages: list[dict]) -> str: |
| task = (re.search(r"\[GÖREV:([A-Z_]+)\]", system) or [None, "DIGER"])[1] |
| user = messages[-1]["content"] if messages else "" |
| en = _lang(system) == "en" |
| if task == "SQL": |
| return _sql_reply(messages) |
| if task == "PING": |
| return "ok" |
| if task == "SINAMA": |
| return "SELECT SUM(tutar) FROM satis WHERE yil = 2024;" |
| if task == "TABLO_ZENGINLESTIR": |
| d = json.loads(user) |
| cap = re.sub(r"^\s*Tablo\s*\d+\s*:\s*", "", d.get("belgedeki_baslik", "")) |
| cols = {} |
| warns = [] |
| for c in d["sutunlar"]: |
| desc = (f"Values of the '{c['orijinal']}' column" if en else f"'{c['orijinal']}' başlıklı sütunun değerleri") |
| cols[c["ad"]] = {"aciklama": desc, "birim": c.get("birim") or ""} |
| if c.get("ayristirilamayan"): |
| warns.append((f"{c['ad']}: {c['ayristirilamayan']} cells could not be parsed as numbers." if en else |
| f"{c['ad']}: {c['ayristirilamayan']} hücre sayı olarak ayrıştırılamadı.")) |
| if c["tip"] == "TEXT" and re.search(r"(19|20)\d{2}", c["ad"]): |
| warns.append(f"{c['ad']}: " + ("year-named column stored as text." if en else "yıl adlı sütun metin olarak saklanmış.")) |
| name = slugify(cap, 40) |
| summ = (f"[MOCK] '{cap}' table with {len(cols)} columns." if en else |
| f"[MOCK] '{cap}' tablosu; {len(cols)} sütun içerir.") |
| return json.dumps({"tablo_adi": name, "ozet": summ, "sutunlar": cols, "uyarilar": warns}, ensure_ascii=False) |
| if task == "SORU_ONER": |
| qs = (["What was İzmir's electricity consumption in 2024?", "Which province consumed the most electricity in 2024?", |
| "How much were personnel expenses in 2024?", "İzmir'in 2024 elektrik tüketimi ne kadar?", |
| "Yürürlük tarihi 15.03.2025 olan hizmetlerin ortalama ücreti nedir?", |
| "Mersin'in 2024 elektrik tüketimi ne kadar?"] if en else |
| ["Faiz giderlerinin bütçesi 2023'ten 2024'e yüzde kaç arttı?", "Marmara bölgesindeki illerin 2024 toplam tüketimi ne kadar?", |
| "Yürürlük tarihi 15.03.2025 olan hizmetlerin ortalama ücreti nedir?", "En fazla su tüketen mahalle hangisi?", |
| "2024 yılında hasılat ne kadardı?", "Mersin'in 2024 elektrik tüketimi ne kadar?"]) |
| return json.dumps(qs, ensure_ascii=False) |
| if task == "CEVAP": |
| m = re.search(r"Biçimlenmiş kısa cevap: (.+)", user) |
| short = m.group(1).strip() if m else "" |
| short = re.sub(r"^(Cevap|Answer): ", "", short) |
| return (f"[MOCK] Based on the tables, the answer is {short}." if en else |
| f"[MOCK] Tablolara göre sorunuzun cevabı: {short}.") |
| if task == "HATA_ACIKLA": |
| d = json.loads(user) |
| if en: |
| return (f"**[MOCK] What happened?** A `{d.get('tip')}` error at the {d.get('asama')} stage.\n\n" |
| f"**Why?** {d.get('mesaj')}\n\n**How to fix?** {'; '.join(d.get('oneriler') or []) or '-'}") |
| return (f"**[MOCK] Ne oldu?** {d.get('asama')} aşamasında `{d.get('tip')}` hatası.\n\n" |
| f"**Neden?** {d.get('mesaj')}\n\n**Nasıl düzeltilir?** {'; '.join(d.get('oneriler') or []) or '-'}") |
| if task == "DEGERLENDIRME_OZETI": |
| rows = json.loads(user) |
| cnt = {} |
| for r in rows: |
| cnt.setdefault(r.get("hata_tipi") or "?", []).append(r["id"]) |
| head = "**[MOCK] Failure themes**" if en else "**[MOCK] Hata temaları**" |
| return head + "\n\n" + "\n".join(f"- **{k}** — {len(v)} ({', '.join(v[:6])})" for k, v in cnt.items()) |
| if task == "TABLO_SEC": |
| names = re.findall(r"^- ([a-z0-9_]+):", user, re.M) |
| q = (re.search(r"^Soru: (.+)$", user, re.M) or [None, ""])[1] |
| try: |
| r = RuleBasedTextToSQL(connect(DB, readonly=True)).generate(q) |
| ranked = [n for _, n in r.trace.get("tablo_puanlari", []) if n in names][:3] |
| except Exception: |
| ranked = [] |
| return json.dumps(ranked or names[:3]) |
| if task == "TABLO_OZETI": |
| return "[MOCK] This table summarises the values in the document." if en else \ |
| "[MOCK] Bu tablo belgedeki değerleri özetler." |
| return "ok" |
|
|
|
|
| class H(BaseHTTPRequestHandler): |
| def log_message(self, *a): |
| pass |
|
|
| def _json(self, code: int, obj, ctype="application/json"): |
| body = json.dumps(obj, ensure_ascii=False).encode() |
| self.send_response(code) |
| self.send_header("Content-Type", ctype) |
| self.send_header("Content-Length", str(len(body))) |
| self.end_headers() |
| self.wfile.write(body) |
|
|
| def do_GET(self): |
| p = self.path.split("?")[0] |
| if p == "/v1/models": |
| if self.headers.get("x-api-key") is not None: |
| if self.headers.get("x-api-key") != "test": |
| return self._json(401, {"type": "error", "error": {"type": "authentication_error", |
| "message": "invalid x-api-key"}}) |
| return self._json(200, {"data": [ |
| {"type": "model", "id": "claude-mock-large", "display_name": "Claude Mock Large", "created_at": "2026-06-01T00:00:00Z"}, |
| {"type": "model", "id": "claude-mock-small", "display_name": "Claude Mock Small", "created_at": "2025-10-01T00:00:00Z"}] |
| + ([{"type": "model", "id": "claude-mock-dengesiz", "display_name": "Claude Mock Dengesiz", |
| "created_at": "2025-01-01T00:00:00Z"}] if EXTRA else []), |
| "has_more": False, "first_id": "claude-mock-large", "last_id": "claude-mock-small"}) |
| return self._json(200, {"object": "list", "data": [ |
| {"id": "gpt-mock-mini", "object": "model", "created": 1750000000, "owned_by": "mock"}, |
| {"id": "gpt-mock", "object": "model", "created": 1760000000, "owned_by": "mock"}] |
| + ([{"id": "gpt-mock-bozuk", "object": "model", "created": 1700000000, "owned_by": "mock"}] if EXTRA else [])}) |
| if p == "/api/version": |
| return self._json(200, {"version": "0.0-mock"}) |
| if p == "/mock/last": |
| return self._json(200, LAST) |
| if p == "/api/tags": |
| return self._json(200, {"models": [{"name": n, "model": n, "size": 4_700_000_000, |
| "details": {"parameter_size": "7B", "quantization_level": "Q4_K_M"}} |
| for n in OLLAMA_MODELS]}) |
| return self._json(404, {"error": "not found"}) |
|
|
| def do_POST(self): |
| n = int(self.headers.get("Content-Length") or 0) |
| d = json.loads(self.rfile.read(n) or b"{}") |
| p = self.path.split("?")[0] |
| if p == "/v1/messages": |
| if self.headers.get("x-api-key") != "test": |
| return self._json(401, {"type": "error", "error": {"type": "authentication_error", "message": "invalid x-api-key"}}) |
| if d.get("model") == "claude-mock-dengesiz" and "[GÖREV:SQL]" in d.get("system", ""): |
| |
| return self._json(529, {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}) |
| txt = reply(d.get("system", ""), d.get("messages", [])) |
| return self._json(200, {"id": "msg_mock", "type": "message", "role": "assistant", "model": d.get("model"), |
| "content": [{"type": "text", "text": txt}], "stop_reason": "end_turn", |
| "usage": {"input_tokens": len(json.dumps(d)) // 4, "output_tokens": len(txt) // 4}}) |
| if p == "/v1/chat/completions": |
| if d.get("model") == "gpt-mock-bozuk": |
| return self._json(400, {"error": {"message": "The model `gpt-mock-bozuk` does not exist", "type": "invalid_request_error"}}) |
| msgs = d.get("messages", []) |
| system = msgs[0]["content"] if msgs and msgs[0]["role"] == "system" else "" |
| txt = reply(system, [m for m in msgs if m["role"] != "system"]) |
| return self._json(200, {"id": "chatcmpl-mock", "object": "chat.completion", "model": d.get("model"), |
| "choices": [{"index": 0, "message": {"role": "assistant", "content": txt}, |
| "finish_reason": "stop"}], |
| "usage": {"prompt_tokens": len(json.dumps(d)) // 4, "completion_tokens": len(txt) // 4}}) |
| if p == "/api/chat": |
| msgs = d.get("messages", []) |
| LAST.clear() |
| LAST.update({"model": d.get("model"), "options": d.get("options", {})}) |
| if d.get("model") not in OLLAMA_MODELS: |
| return self._json(404, {"error": f"model '{d.get('model')}' not found"}) |
| system = msgs[0]["content"] if msgs and msgs[0]["role"] == "system" else "" |
| txt = reply(system, [m for m in msgs if m["role"] != "system"]) |
| return self._json(200, {"model": d.get("model"), "message": {"role": "assistant", "content": txt}, |
| "done": True, "prompt_eval_count": len(json.dumps(d)) // 4, |
| "eval_count": len(txt) // 4}) |
| if p == "/api/pull": |
| name = d.get("model") or d.get("name") |
| self.send_response(200) |
| self.send_header("Content-Type", "application/x-ndjson") |
| self.end_headers() |
| for line in ({"status": "pulling manifest"}, {"status": "downloading", "completed": 50, "total": 100}, |
| {"status": "downloading", "completed": 100, "total": 100}, {"status": "success"}): |
| self.wfile.write((json.dumps(line) + "\n").encode()) |
| self.wfile.flush() |
| time.sleep(0.05) |
| if name and name not in OLLAMA_MODELS: |
| OLLAMA_MODELS.append(name) |
| return None |
| return self._json(404, {"error": "not found"}) |
|
|
|
|
| def serve(port: int = 8765): |
| srv = ThreadingHTTPServer(("127.0.0.1", port), H) |
| srv.serve_forever() |
|
|
|
|
| if __name__ == "__main__": |
| serve(int(sys.argv[1]) if len(sys.argv) > 1 else 8765) |
|
|
