- LLMProvider: complete(messages, schema) -> LLMResult (текст, разобранный JSON, токены) - openai_compat: /chat/completions, structured output через response_format json_schema, JSON из ```-обёртки тоже разбирается; ретраи на 429/5xx и сетевые ошибки, 4xx — сразу ошибка - mock: детерминированные правила для тестов и демо без сети - LLMSettings (LLM_PROVIDER/BASE_URL/MODEL/API_KEY…), CLI python -m app.llm.cli Refs #17 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
37 lines
1.4 KiB
Python
37 lines
1.4 KiB
Python
"""Deterministic offline provider for tests and demos without network or tokens.
|
|
|
|
Rules are checked in order: the first whose `contains` is found in the last user message wins.
|
|
Without a match it echoes the message (or returns `{}` when JSON was requested).
|
|
"""
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from app.llm.provider import ChatMessage, JsonSchema, LLMResult
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MockRule:
|
|
contains: str
|
|
reply: str | dict[str, Any]
|
|
|
|
|
|
class MockProvider:
|
|
def __init__(self, rules: list[MockRule] | None = None) -> None:
|
|
self.rules = list(rules or [])
|
|
self.calls: list[tuple[list[ChatMessage], JsonSchema | None]] = []
|
|
|
|
async def aclose(self) -> None:
|
|
pass
|
|
|
|
async def complete(self, messages: list[ChatMessage], schema: JsonSchema | None = None) -> LLMResult:
|
|
self.calls.append((messages, schema))
|
|
last = next((m.content for m in reversed(messages) if m.role == "user"), "")
|
|
reply: str | dict[str, Any] | None = next((r.reply for r in self.rules if r.contains in last), None)
|
|
if reply is None:
|
|
reply = {} if schema is not None else f"mock: {last}"
|
|
text = json.dumps(reply, ensure_ascii=False) if isinstance(reply, dict) else reply
|
|
data = reply if isinstance(reply, dict) and schema is not None else None
|
|
return LLMResult(text=text, data=data, prompt_tokens=len(last.split()), completion_tokens=len(text.split()))
|