synthos/cli.py

"""Synthos CLI.  Usage:

    python -m synthos ask "your question"        one-shot, with memory + refine
    python -m synthos ask "..." --image pic.png  multimodal: reason over an image + text
    python -m synthos ask "..." --audio clip.wav multimodal: transcribe speech, then reason
    python -m synthos see pic.png ["question"]   describe/answer about an image (open VLM)
    python -m synthos hear clip.wav ["question"] transcribe speech, then reason with memory
    python -m synthos chat                         interactive session (persists memory)
    python -m synthos remember "a durable fact"  store a fact
    python -m synthos recall "cue"               show what memory surfaces for a cue
    python -m synthos stats                        memory + model status

Multimodal = routing to OPEN backends (LLaVA/Qwen-VL for vision, Whisper for audio) and
fusing their output with memory + reasoning. Not a from-scratch multimodal model.

Flags: --profile {laptop,commander,phone}  --think {refine,vote}  --adaptive  --route
       --k N  --rounds N  --n N
       --image PATH  --audio PATH  --vision-expert KEY  --whisper-model SIZE
"""
from __future__ import annotations

import argparse
import json
import sys

from .engine import Synthos


def _bot(args) -> Synthos:
    return Synthos(profile=args.profile, think=args.think,
                   adaptive=getattr(args, "adaptive", False),
                   route=getattr(args, "route", False),
                   vision_expert=getattr(args, "vision_expert", "vision"),
                   whisper_model=getattr(args, "whisper_model", "base.en"))


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(prog="synthos")
    ap.add_argument("--profile", default="laptop",
                    choices=["laptop", "commander", "phone"])
    ap.add_argument("--think", default="refine", choices=["refine", "vote"])
    ap.add_argument("--adaptive", action="store_true",
                    help="spend test-time compute proportional to difficulty (recommended)")
    ap.add_argument("--route", action="store_true",
                    help="route each task to its best-fit local expert (model-level MoE)")
    ap.add_argument("--k", type=int, default=4, help="memories to recall")
    ap.add_argument("--rounds", type=int, default=1, help="refine critique/revise rounds")
    ap.add_argument("--n", type=int, default=1, help="vote samples (think=vote)")
    ap.add_argument("--image", default=None, help="image path to reason over (multimodal)")
    ap.add_argument("--audio", default=None, help="audio path to transcribe (multimodal)")
    ap.add_argument("--vision-expert", dest="vision_expert", default="vision",
                    help="router expert key for the VLM (default: vision -> llava)")
    ap.add_argument("--whisper-model", dest="whisper_model", default="base.en",
                    help="faster-whisper model size (default: base.en)")
    sub = ap.add_subparsers(dest="cmd", required=True)
    a = sub.add_parser("ask"); a.add_argument("query", nargs="*")
    s = sub.add_parser("see"); s.add_argument("image"); s.add_argument("query", nargs="*")
    h = sub.add_parser("hear"); h.add_argument("audio"); h.add_argument("query", nargs="*")
    sub.add_parser("chat")
    r = sub.add_parser("remember"); r.add_argument("fact", nargs="+")
    rc = sub.add_parser("recall"); rc.add_argument("cue", nargs="+")
    sub.add_parser("stats")
    args = ap.parse_args(argv)

    if args.cmd == "stats":
        print(json.dumps(_bot(args).stats(), indent=2)); return 0
    if args.cmd == "remember":
        bot = _bot(args)
        print("stored:", bot.remember(" ".join(args.fact))); return 0
    if args.cmd == "recall":
        for r_ in _bot(args).recall(" ".join(args.cue), k=args.k):
            print(f"  [{r_.score:+.3f}] ({r_.kind}) {r_.summary}")
        return 0
    if args.cmd in ("ask", "see", "hear"):
        n = args.n if args.think == "vote" else 1
        image = args.image
        audio = args.audio
        if args.cmd == "see":
            image = args.image
        if args.cmd == "hear":
            audio = args.audio
        text = " ".join(args.query)
        out = _bot(args).ask(text, k=args.k, rounds=args.rounds, n=n,
                             image=image, audio=audio)
        print(out["answer"])
        if out.get("modality"):
            print("\n--", json.dumps(out["modality"]), file=sys.stderr)
        print("\n--", json.dumps(out["usage"]), file=sys.stderr)
        return 0
    if args.cmd == "chat":
        bot = _bot(args)
        print(f"Synthos [{bot.client.p.name}: {bot.client.p.model}] — Ctrl-C to exit.")
        n = args.n if args.think == "vote" else 1
        try:
            while True:
                q = input("\nyou> ").strip()
                if not q:
                    continue
                out = bot.ask(q, k=args.k, rounds=args.rounds, n=n)
                print(f"\nsynthos> {out['answer']}")
        except (KeyboardInterrupt, EOFError):
            print("\nbye."); return 0
    return 1


if __name__ == "__main__":
    raise SystemExit(main())