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>
50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
"""Provider and task protocols of the LLM layer.
|
|
|
|
Two levels, so the offline mock does not have to reverse-engineer prompts:
|
|
|
|
* `LLMProvider` — transport: "send system prompt + scrubbed user text, get JSON/text back".
|
|
Implemented by GigaChat (`giga.py`). `PromptedTasks` (`tasks.py`) turns it into tasks.
|
|
* `LLMTasks` — task level: extract_item / extract_checkin / risk / phrase / summarize.
|
|
Implemented by `PromptedTasks(provider)` and by the rule-based `MockTasks` (`mock.py`).
|
|
|
|
The gateway (`gateway.py`) wraps an `LLMTasks` with the shared policy: scrub guard, confidence
|
|
threshold, forbidden-output filter, template fallbacks. Tasks may return `None` on any failure.
|
|
"""
|
|
|
|
from typing import Any, Protocol, TypeVar
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from app.core.types import ScrubbedText
|
|
from app.llm.schemas import CheckinFacts, ItemScore, RiskVerdict
|
|
|
|
M = TypeVar("M", bound=BaseModel)
|
|
|
|
|
|
class LLMProvider(Protocol):
|
|
async def complete_json(
|
|
self,
|
|
system: str,
|
|
user: ScrubbedText,
|
|
schema: type[M],
|
|
validation_context: dict[str, Any] | None = None,
|
|
) -> M | None: ...
|
|
|
|
async def complete_text(self, system: str, user: ScrubbedText) -> str | None: ...
|
|
|
|
|
|
class LLMTasks(Protocol):
|
|
name: str
|
|
|
|
async def extract_item(
|
|
self, scale_item_id: str, item_text: str, max_score: int, answer: ScrubbedText
|
|
) -> ItemScore | None: ...
|
|
|
|
async def extract_checkin(self, text: ScrubbedText) -> CheckinFacts | None: ...
|
|
|
|
async def risk_classify(self, text: ScrubbedText) -> RiskVerdict | None: ...
|
|
|
|
async def phrase(self, intent: str, payload: ScrubbedText, base_text: str) -> str | None: ...
|
|
|
|
async def summarize_for_doctor(self, structured: ScrubbedText, base_text: str) -> str | None: ...
|