#!/usr/bin/env python3 """TON-2936 Lab demo #1: constrained decoding experiment. Runs the same JSON-extraction prompts against Qwen3.6-27B (Atlas :8020) in two modes — prompt-only vs grammar-enforced (json_schema) — and records raw outputs + validity for the static demo page. """ import json, time, urllib.request, sys API = "http://192.168.1.6:8020/v1/chat/completions" MODEL = "Qwen3.6-27B-Q4_K_M.gguf" OUT = "/Volumes/Data/tony_ai_lab/codennect/lab/constrained-decoding/results.json" import jsonschema TASKS = [ { "id": "event", "name": "일정 추출", "desc": "카톡/문자에서 일정 정보를 구조화", "schema": { "type": "object", "properties": { "title": {"type": "string"}, "date": {"type": "string", "pattern": "^\\d{4}-\\d{2}-\\d{2}$"}, "start_time": {"type": ["string", "null"]}, "location": {"type": ["string", "null"]}, "attendees": {"type": "array", "items": {"type": "string"}}, }, "required": ["title", "date", "start_time", "location", "attendees"], "additionalProperties": False, }, "instruction": "다음 메시지에서 일정 정보를 추출해 JSON으로 출력하세요. 기준 연도는 2026년입니다. 정보가 없으면 null을 쓰세요.", "inputs": [ "다음주 금요일 7시에 강남역 3번출구 앞에서 민수, 지혜랑 저녁 어때? 7월 17일이야!", "[Web발신] 7/21(화) 14:00 삼성서울병원 본관 3층 내과 외래 예약이 확정되었습니다.", "야 우리 팀 회식 잡혔다 ㅋㅋ 8월 1일 토요일 저녁 6시반 을지로 골뱅이집. 참석자: 나, 너, 팀장님, 대리님", "리마인드: 분기 전략 리뷰는 7월 30일 오전 10시, 본사 12층 대회의실입니다. 발표자는 김태호 님과 이서연 님입니다.", "낼모레 등산 가는거 알지? 새벽 5시 사당역 집결! (오늘은 7월 8일)", ], }, { "id": "review", "name": "리뷰 분석", "desc": "쇼핑몰 리뷰의 감성·장단점 구조화", "schema": { "type": "object", "properties": { "sentiment": {"type": "string", "enum": ["positive", "negative", "mixed"]}, "score": {"type": "integer", "minimum": 1, "maximum": 5}, "pros": {"type": "array", "items": {"type": "string"}}, "cons": {"type": "array", "items": {"type": "string"}}, "one_line": {"type": "string"}, }, "required": ["sentiment", "score", "pros", "cons", "one_line"], "additionalProperties": False, }, "instruction": "다음 상품 리뷰를 분석해 JSON으로 출력하세요. score는 리뷰어가 매겼을 법한 별점(1~5 정수)입니다.", "inputs": [ "배송은 진짜 빨랐어요. 근데 화면에 죽은 픽셀이 하나 있네요... 교환하기는 귀찮고 그냥 씁니다. 가격 생각하면 뭐 그럭저럭.", "인생 이어폰입니다. 노캔 미쳤고 착용감도 좋아요. 케이스가 지문이 잘 묻는 것만 빼면 완벽!", "최악. 한 달 만에 충전이 안 됨. 고객센터는 전화도 안 받고. 절대 사지 마세요.", "무난합니다. 딱 가격만큼 하는 제품. 기대도 실망도 없음.", "선물용으로 샀는데 포장이 너무 허접해서 당황;; 제품 자체는 받는 분이 만족하셨어요. 향도 좋다고 하시고요.", ], }, { "id": "meeting", "name": "회의록 액션아이템", "desc": "회의록에서 결정사항·할 일 구조화", "schema": { "type": "object", "properties": { "summary": {"type": "string"}, "decisions": {"type": "array", "items": {"type": "string"}}, "action_items": { "type": "array", "items": { "type": "object", "properties": { "task": {"type": "string"}, "owner": {"type": "string"}, "due": {"type": ["string", "null"]}, }, "required": ["task", "owner", "due"], "additionalProperties": False, }, }, }, "required": ["summary", "decisions", "action_items"], "additionalProperties": False, }, "instruction": "다음 회의록을 요약하고 결정사항과 액션아이템을 JSON으로 출력하세요. 마감일이 없으면 null을 쓰세요.", "inputs": [ "오늘 스프린트 회고. 배포 자동화가 계속 밀리는 문제 논의. 결론: CI 파이프라인 개선을 다음 스프린트 최우선으로. 준호가 금요일까지 초안 PR, 수아가 리뷰 담당.", "마케팅 주간회의. 7월 캠페인 성과 CTR 2.1%로 목표 미달. A/B 테스트 확대하기로 결정. 카피 시안 3종은 다은이 7/15까지, 랜딩 개선은 개발팀에 요청(담당 미정).", "장애 포스트모템: 새벽 2시 DB 커넥션 풀 고갈로 30분 다운. 원인은 배치 잡의 커넥션 누수. 재발 방지로 커넥션 모니터링 알람 추가(태호, 이번 주), 배치 잡 코드 리뷰(서연, 7월 말까지).", "신규 입사자 온보딩 개선 회의. 문서가 흩어져 있다는 피드백 다수. 위키 통합은 하기로 했고, 멘토링 제도는 다음 분기에 다시 논의. 위키 정리는 현우 담당, 기한은 미정.", "짧은 싱크. 다음 주 데모데이 리허설 일정만 확정: 수요일 오후 3시. 데모 시나리오 준비는 각자. 특이사항 없음.", ], }, ] SYSTEM = "당신은 정보 추출 엔진입니다. 반드시 유효한 JSON 객체 하나만 출력하세요. 마크다운 코드펜스, 설명, 그 외 어떤 텍스트도 붙이지 마세요." def call(messages, schema=None, max_tokens=600): body = { "model": MODEL, "messages": messages, "temperature": 0.7, "max_tokens": max_tokens, "chat_template_kwargs": {"enable_thinking": False}, } if schema is not None: body["json_schema"] = schema req = urllib.request.Request(API, json.dumps(body).encode(), {"Content-Type": "application/json"}) t0 = time.time() with urllib.request.urlopen(req, timeout=600) as r: d = json.load(r) dt = time.time() - t0 return d["choices"][0]["message"]["content"], dt def lenient_extract(raw): s = raw.strip() if s.startswith("```"): s = s.split("```", 2)[1] if s.startswith("json"): s = s[4:] s = s.strip() a, b = s.find("{"), s.rfind("}") if a != -1 and b > a: s = s[a : b + 1] return s def validate(raw, schema): out = {"strict_parse": False, "lenient_parse": False, "schema_valid": False, "error": None} obj = None try: obj = json.loads(raw) out["strict_parse"] = True out["lenient_parse"] = True except Exception as e: out["error"] = f"strict parse: {e}" try: obj = json.loads(lenient_extract(raw)) out["lenient_parse"] = True except Exception as e2: out["error"] = f"parse failed even leniently: {e2}" return out try: jsonschema.validate(obj, schema) out["schema_valid"] = True except jsonschema.ValidationError as e: out["error"] = f"schema: {e.message}" return out def main(): results = [] total = sum(len(t["inputs"]) for t in TASKS) * 2 n = 0 for task in TASKS: for i, inp in enumerate(task["inputs"]): msgs = [ {"role": "system", "content": SYSTEM}, {"role": "user", "content": f"{task['instruction']}\n\nJSON 스키마:\n{json.dumps(task['schema'], ensure_ascii=False)}\n\n입력:\n{inp}"}, ] for mode in ("free", "constrained"): n += 1 schema = task["schema"] if mode == "constrained" else None try: raw, dt = call(msgs, schema) except Exception as e: print(f"[{n}/{total}] {task['id']}#{i} {mode} REQUEST FAILED: {e}", flush=True) results.append({"task": task["id"], "input_idx": i, "mode": mode, "raw": None, "latency_s": None, "request_error": str(e)}) continue v = validate(raw, task["schema"]) results.append({"task": task["id"], "input_idx": i, "mode": mode, "raw": raw, "latency_s": round(dt, 2), **v}) print(f"[{n}/{total}] {task['id']}#{i} {mode}: strict={v['strict_parse']} schema={v['schema_valid']} {round(dt,1)}s", flush=True) meta = {"model": MODEL, "temperature": 0.7, "date": "2026-07-07", "runs": results, "tasks": [{"id": t["id"], "name": t["name"], "desc": t["desc"], "schema": t["schema"], "instruction": t["instruction"], "inputs": t["inputs"]} for t in TASKS]} with open(OUT, "w") as f: json.dump(meta, f, ensure_ascii=False, indent=1) print(f"wrote {OUT} ({len(results)} runs)") if __name__ == "__main__": main()