| """Fallback interactive CLI chat (PRIVATE to Serving_Module) - used when TELEGRAM_BOT_TOKEN is empty. |
| |
| Commands |
| mode thinking | mode non_thinking | mode max_thinking switch the answer mode |
| teach: सवाल | जवाब teach the model a fact right now |
| clear forget the conversation so far |
| help show this list |
| exit | quit leave |
| """ |
| from __future__ import annotations |
|
|
| import re |
| import sys |
|
|
| from .inference_engine import MODES, InferenceEngine, normalise_mode |
|
|
| HELP = __doc__.split("Commands", 1)[1].strip("\n") |
|
|
|
|
| def _split_teach(text: str): |
| body = text.split(":", 1)[1] if ":" in text else "" |
| if "|" not in body: |
| return None, None |
| question, _, answer = body.partition("|") |
| return question.strip(), answer.strip() |
|
|
|
|
| def _read_line(out, mode: str) -> str: |
| out.write(f"\n[{mode}] आप> ") |
| out.flush() |
| line = sys.stdin.readline() |
| if line == "": |
| raise EOFError |
| return line |
|
|
|
|
| def run_cli(engine: InferenceEngine, start_mode: str = "non_thinking", stream_out=None) -> None: |
| out = stream_out or sys.stdout |
| mode = normalise_mode(start_mode) or "non_thinking" |
|
|
| def say(text: str = "", end: str = "\n") -> None: |
| out.write(text + end) |
| out.flush() |
|
|
| say("\n" + "=" * 68) |
| say(" RSS AI - interactive chat (type 'help' for commands, 'exit' to leave)") |
| say(f" current mode: {mode} | modes: {', '.join(MODES)}") |
| say("=" * 68) |
| while True: |
| try: |
| line = input(f"\n[{mode}] आप> ") if stream_out is None else _read_line(out, mode) |
| except EOFError: |
| say("\n(input closed - bye)") |
| return |
| except KeyboardInterrupt: |
| say("\nbye!") |
| return |
| text = line.strip() |
| if not text: |
| continue |
| low = text.lower() |
| if low in ("exit", "quit", "/exit", "/quit", "q"): |
| say("bye!") |
| return |
| if low in ("help", "/help", "?"): |
| say(HELP) |
| continue |
| if low in ("clear", "/clear"): |
| engine.reset() |
| say("conversation cleared.") |
| continue |
| if low.startswith(("mode ", "/mode ")): |
| chosen = normalise_mode(text.split(None, 1)[1]) |
| if chosen: |
| mode = chosen |
| say(f"mode -> {mode}") |
| else: |
| say(f"unknown mode. choose one of: {', '.join(MODES)}") |
| continue |
| if low.startswith("/") and low.lstrip("/") in MODES: |
| mode = low.lstrip("/") |
| say(f"mode -> {mode}") |
| continue |
| if re.match(r"teach\s*:", low): |
| question, answer = _split_teach(text) |
| if question is None: |
| say("format: teach: सवाल | जवाब") |
| else: |
| say(engine.teach(question, answer)) |
| continue |
| try: |
| say("RSS AI> ", end="") |
| result = engine.generate(text, mode=mode, on_text=lambda chunk: say(chunk, end="")) |
| say(f"\n ({result['mode']}, {result['tokens']} tokens, {result['seconds']}s, from {result['source']})") |
| except KeyboardInterrupt: |
| say("\n (generation interrupted)") |
| except Exception as exc: |
| say(f"\n error: {type(exc).__name__}: {exc}") |
