|
| 1 | +"""L3.5 · 확정/논의 게이트 (AI-05). |
| 2 | +
|
| 3 | +L4(`l4.py`)를 복제해 세 곳만 바꾼 것이다 — SPEC · build_response_schema · 후처리. |
| 4 | +
|
| 5 | +**precision 우선을 프롬프트가 아니라 후처리로 강제한다.** 프롬프트의 "애매하면 |
| 6 | +DISCUSSED" 는 지켜지지 않을 확률이 남지만, 여기서는 그 확률이 남는 자리를 없앤다: |
| 7 | +
|
| 8 | + 모델이 판정을 돌려주지 않은 항목 → DISCUSSED |
| 9 | +
|
| 10 | +즉 **누락이 통과가 되지 않는다.** 반대로 두면(누락 = CONFIRMED) 응답이 잘리거나 |
| 11 | +항목 하나를 건너뛴 것만으로 합의되지 않은 일이 배정으로 나간다. 게이트에서 가장 |
| 12 | +비싼 실패는 그것이다. |
| 13 | +""" |
| 14 | + |
| 15 | +from __future__ import annotations |
| 16 | + |
| 17 | +from app.layers import few_shot |
| 18 | +from app.layers import formatting as fmt |
| 19 | +from app.layers.runner import LayerRunner, LayerSpec |
| 20 | +from app.schemas.common import FewShotExample |
| 21 | +from app.schemas.l3_5 import ( |
| 22 | + REASON_MAX, |
| 23 | + GateCandidate, |
| 24 | + GateRequest, |
| 25 | + GateResponse, |
| 26 | + GateVerdict, |
| 27 | +) |
| 28 | + |
| 29 | +SPEC = LayerSpec( |
| 30 | + layer="L3.5", |
| 31 | + prompt_file="l3_5_gate.v1.txt", |
| 32 | + prompt_version="v1", |
| 33 | + dry_run_payload={"verdicts": []}, |
| 34 | +) |
| 35 | + |
| 36 | +CONFIRMED = "CONFIRMED" |
| 37 | +DISCUSSED = "DISCUSSED" |
| 38 | +GATE_STATUSES = [CONFIRMED, DISCUSSED] |
| 39 | + |
| 40 | +# 모델이 답하지 않은 항목에 붙는 사유. 사람이 화면에서 이 문구를 보면 "게이트가 |
| 41 | +# 판정을 못 했다"와 "게이트가 논의로 판정했다"를 구분할 수 있어야 한다. |
| 42 | +MISSING_VERDICT_REASON = "게이트가 이 항목을 판정하지 않았다 — 확정으로 올리지 않는다(precision 우선)." |
| 43 | + |
| 44 | + |
| 45 | +def build_response_schema(items: list[GateCandidate]) -> dict: |
| 46 | + """itemKey 를 요청 항목의 enum 으로 박는다 — 없는 항목에 대한 판정이 나올 수 없게.""" |
| 47 | + return { |
| 48 | + "type": "OBJECT", |
| 49 | + "properties": { |
| 50 | + "verdicts": { |
| 51 | + "type": "ARRAY", |
| 52 | + "items": { |
| 53 | + "type": "OBJECT", |
| 54 | + "properties": { |
| 55 | + "itemKey": {"type": "STRING", "enum": [item.item_key for item in items]}, |
| 56 | + "gateStatus": {"type": "STRING", "enum": GATE_STATUSES}, |
| 57 | + "reason": {"type": "STRING"}, |
| 58 | + }, |
| 59 | + "required": ["itemKey", "gateStatus", "reason"], |
| 60 | + "propertyOrdering": ["itemKey", "gateStatus", "reason"], |
| 61 | + }, |
| 62 | + } |
| 63 | + }, |
| 64 | + "required": ["verdicts"], |
| 65 | + } |
| 66 | + |
| 67 | + |
| 68 | +async def gate(request: GateRequest, runner: LayerRunner) -> GateResponse: |
| 69 | + examples = await few_shot.lookup( |
| 70 | + tenant_id=request.tenant_id, |
| 71 | + layer=SPEC.layer, |
| 72 | + query_text=request.query_text, |
| 73 | + ) |
| 74 | + |
| 75 | + if not request.items: |
| 76 | + return GateResponse( |
| 77 | + verdicts=[], |
| 78 | + used_few_shot=examples, |
| 79 | + model=runner.model_name, |
| 80 | + prompt_version=SPEC.prompt_version, |
| 81 | + ) |
| 82 | + |
| 83 | + schema = build_response_schema(request.items) |
| 84 | + variables = build_prompt_variables(request, examples) |
| 85 | + |
| 86 | + raw, usage = await runner.run(SPEC, variables=variables, response_schema=schema) |
| 87 | + |
| 88 | + return GateResponse( |
| 89 | + verdicts=parse_verdicts(raw, request.items), |
| 90 | + used_few_shot=examples, |
| 91 | + usage=usage, |
| 92 | + model=runner.model_name, |
| 93 | + prompt_version=SPEC.prompt_version, |
| 94 | + ) |
| 95 | + |
| 96 | + |
| 97 | +def build_prompt_variables(request: GateRequest, examples: list[FewShotExample]) -> dict[str, str]: |
| 98 | + return { |
| 99 | + "TOPIC": request.topic, |
| 100 | + "PARTICIPANTS": fmt.format_participants(request.participants), |
| 101 | + "ITEMS": _format_items(request.items), |
| 102 | + "UTTERANCES": fmt.format_utterances(request.utterances), |
| 103 | + "FEW_SHOT": fmt.format_few_shot(examples), |
| 104 | + } |
| 105 | + |
| 106 | + |
| 107 | +def parse_verdicts(raw: dict, items: list[GateCandidate]) -> list[GateVerdict]: |
| 108 | + """**요청 항목 수만큼 정확히 반환한다.** 판정이 없는 항목은 DISCUSSED 로 채운다. |
| 109 | +
|
| 110 | + 요청 순서를 유지하는 것도 계약이다 — Spring 이 순서로 맞추지는 않지만, 순서가 |
| 111 | + 흔들리면 응답을 눈으로 대조할 때 사람이 항목을 놓친다. |
| 112 | + """ |
| 113 | + by_key: dict[str, GateVerdict] = {} |
| 114 | + |
| 115 | + for entry in raw.get("verdicts") or []: |
| 116 | + if not isinstance(entry, dict): |
| 117 | + continue |
| 118 | + key = fmt.as_text(entry.get("itemKey")) |
| 119 | + status = entry.get("gateStatus") |
| 120 | + # 목록 밖 키, 알 수 없는 판정값은 버린다 — 아래에서 DISCUSSED 로 채워진다. |
| 121 | + if not key or status not in GATE_STATUSES: |
| 122 | + continue |
| 123 | + if key in by_key: # 같은 항목을 두 번 판정했다. 먼저 온 것만 쓴다 |
| 124 | + continue |
| 125 | + by_key[key] = GateVerdict( |
| 126 | + item_key=key, |
| 127 | + gate_status=status, |
| 128 | + reason=fmt.clip(fmt.as_text(entry.get("reason")), REASON_MAX) |
| 129 | + or "(모델이 판정 근거를 반환하지 않음)", |
| 130 | + ) |
| 131 | + |
| 132 | + return [ |
| 133 | + by_key.get( |
| 134 | + item.item_key, |
| 135 | + GateVerdict(item_key=item.item_key, gate_status=DISCUSSED, reason=MISSING_VERDICT_REASON), |
| 136 | + ) |
| 137 | + for item in items |
| 138 | + ] |
| 139 | + |
| 140 | + |
| 141 | +def _format_items(items: list[GateCandidate]) -> str: |
| 142 | + lines = [ |
| 143 | + f"- itemKey={item.item_key} · [{item.item_type}] {item.content}" |
| 144 | + f" (근거 발화 id={item.evidence_utterance_id})" |
| 145 | + for item in items |
| 146 | + ] |
| 147 | + return "\n".join(lines) or fmt.NONE_MARK |
0 commit comments