ElebobaandClaude Opus 5.5
924a51c483
CI / ruff · pytest (pull_request) Successful in 23s
Single entry point app/llm/gateway.py: extract_item, extract_checkin, risk_classify, phrase, summarize_for_doctor. Two-level design: LLMProvider (transport, GigaChat) and LLMTasks (PromptedTasks over a provider, or rule-based MockTasks). Shared policy in the gateway: assert_scrubbed backstop, confidence threshold 0.5, risk failure -> NONE, forbidden-output filter with config/texts/phrases.yaml fallbacks. Refs #7 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
"""Manual check of the LLM gateway on synthetic phrases.
|
|
|
|
python -m app.llm.cli extract "сплю плохо, кошмары почти каждую ночь"
|
|
python -m app.llm.cli item phq9_3 "Проблемы со сном" 3 "сплю плохо"
|
|
python -m app.llm.cli risk "всё бессмысленно"
|
|
|
|
Text always goes through `privacy.scrub()` first; the CLI never bypasses it.
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import sys
|
|
|
|
from app.core import privacy
|
|
from app.llm.gateway import get_gateway
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(prog="python -m app.llm.cli")
|
|
sub = parser.add_subparsers(dest="cmd", required=True)
|
|
sub.add_parser("extract", help="daily check-in facts").add_argument("text")
|
|
sub.add_parser("risk", help="auxiliary risk level").add_argument("text")
|
|
item = sub.add_parser("item", help="scale item score")
|
|
item.add_argument("item_id")
|
|
item.add_argument("item_text")
|
|
item.add_argument("max_score", type=int)
|
|
item.add_argument("text")
|
|
args = parser.parse_args(argv)
|
|
|
|
try:
|
|
scrubbed = privacy.scrub(args.text)
|
|
except NotImplementedError:
|
|
msg = "privacy.scrub() is not implemented yet (#2); refusing to send unscrubbed text."
|
|
print(msg, file=sys.stderr)
|
|
return 2
|
|
|
|
gw = get_gateway()
|
|
print(f"provider={gw.provider_name}")
|
|
if args.cmd == "extract":
|
|
facts = asyncio.run(gw.extract_checkin(scrubbed))
|
|
print(facts.model_dump_json(indent=2) if facts else "None")
|
|
elif args.cmd == "risk":
|
|
print(asyncio.run(gw.risk_classify(scrubbed)).name)
|
|
else:
|
|
score = asyncio.run(gw.extract_item(args.item_id, args.item_text, args.max_score, scrubbed))
|
|
print(score.model_dump_json(indent=2) if score else "None (low confidence -> clarify)")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|