অ্যাডাপ্টার, কনফিগারেশন ও এরর হ্যান্ডলিং

# file: evidencedesk/config.py
"""Validated settings. The environment is passed in as a mapping, so tests never touch os.environ."""
from __future__ import annotations

from collections.abc import Mapping
from pathlib import Path
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, SecretStr, model_validator

PREFIX = "EVIDENCEDESK_"
NEEDS_KEY = ("openai", "anthropic", "gemini")  # "local" and "fake" need no key


class Settings(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")
    provider: Literal["fake", "openai", "anthropic", "gemini", "local"] = "fake"
    model: str = "fake-model"
    timeout_s: float = Field(30.0, gt=0, le=600)
    max_retries: int = Field(2, ge=0, le=5)
    api_key: SecretStr | None = None  # prints as ********** in repr and logs

    @model_validator(mode="after")
    def _check_provider(self) -> Settings:
        if self.provider != "fake" and self.model == "fake-model":
            raise ValueError(f"set {PREFIX}MODEL: copy the model id from the provider's documentation")
        if self.provider in NEEDS_KEY and self.api_key is None:
            raise ValueError(f"provider {self.provider!r} needs {PREFIX}API_KEY or {PREFIX}API_KEY_FILE")
        return self


def load_settings(env: Mapping[str, str]) -> Settings:
    """Build Settings from an environment mapping. An empty value counts as not set."""
    data: dict[str, str] = {}
    for field in ("provider", "model", "timeout_s", "max_retries", "api_key"):
        if env.get(PREFIX + field.upper()):
            data[field] = env[PREFIX + field.upper()]
    key_file = env.get(PREFIX + "API_KEY_FILE")
    if "api_key" not in data and key_file:
        data["api_key"] = Path(key_file).read_text(encoding="utf-8").strip()
    return Settings(**data)

from pydantic import ValidationError

from evidencedesk.config import load_settings

print(load_settings({}))
secret = "demo-key-not-real"
ready = load_settings({"EVIDENCEDESK_PROVIDER": "openai", "EVIDENCEDESK_MODEL": "model-id-from-the-docs",
                       "EVIDENCEDESK_API_KEY": secret})
print(ready)
print(secret in repr(ready), ready.api_key.get_secret_value() == secret)
for env in ({"EVIDENCEDESK_TIMEOUT_S": "0"}, {"EVIDENCEDESK_PROVIDER": "gemini", "EVIDENCEDESK_MODEL": "m1"}):
    try:
        load_settings(env)
    except ValidationError as error:
        print([(e["loc"], e["msg"]) for e in error.errors()])

provider='fake' model='fake-model' timeout_s=30.0 max_retries=2 api_key=None
provider='openai' model='model-id-from-the-docs' timeout_s=30.0 max_retries=2 api_key=SecretStr('**********')
False True
[(('timeout_s',), 'Input should be greater than 0')]
[((), "Value error, provider 'gemini' needs EVIDENCEDESK_API_KEY or EVIDENCEDESK_API_KEY_FILE")]

# file: .gitignore
.env
.env.*
!.env.example
*.key
secrets/
.venv/
__pycache__/

# file: evidencedesk/errors.py
from __future__ import annotations

import random
import time
from collections.abc import Callable, Mapping
from dataclasses import dataclass

from evidencedesk.llm import LLMClient, LLMRequest, LLMResponse

RETRYABLE = {"timeout", "rate_limited", "server"}
QUOTA_MARKERS = ("credit_balance_exhausted", "spend_limit", "quota_exceeded")  # our heuristic


class LLMError(Exception):
    def __init__(self, kind: str, message: str = "", retry_after: float | None = None):
        super().__init__(f"{kind}: {message}" if message else kind)
        self.kind, self.retry_after, self.attempts = kind, retry_after, 1

    @property
    def retryable(self) -> bool:
        return self.kind in RETRYABLE


def from_status(status: int, headers: Mapping[str, str] | None = None, body: str = "") -> LLMError:
    lowered = {k.lower(): v for k, v in (headers or {}).items()}
    try:
        retry_after = max(0.0, float(lowered["retry-after"]))  # seconds only; a date is ignored
    except (KeyError, ValueError):
        retry_after = None
    if status == 429 and any(marker in body for marker in QUOTA_MARKERS):
        kind = "quota_exhausted"
    elif status in (408, 504):
        kind = "timeout"
    elif status == 429:
        kind = "rate_limited"
    elif status == 402:
        kind = "quota_exhausted"
    elif status in (401, 403):
        kind = "auth"
    elif status in (400, 404, 413, 422):
        kind = "invalid_request"
    elif status >= 500:
        kind = "server"
    else:
        kind = "unknown"
    return LLMError(kind, f"HTTP {status}", retry_after)


@dataclass(frozen=True)
class RetryPolicy:
    max_attempts: int = 3  # every try counts, the first too
    base_delay: float = 0.5
    factor: float = 2.0
    max_delay: float = 8.0  # the longest wait we accept
    jitter: float = 0.5  # share of each wait that is random

    def wait_for(self, attempt: int, error: LLMError, rng: random.Random) -> float | None:
        backoff = min(self.max_delay, self.base_delay * self.factor ** (attempt - 1))
        wait = backoff * (1 - self.jitter * rng.random())
        wait = max(wait, error.retry_after or 0.0)  # the server's word beats our guess
        return wait if wait <= self.max_delay else None


def call_with_retry(fn: Callable, policy: RetryPolicy = RetryPolicy(),
                    sleep: Callable[[float], None] = time.sleep, rng: random.Random | None = None):
    rng = rng or random.Random()
    for attempt in range(1, policy.max_attempts + 1):
        try:
            return fn()
        except LLMError as error:
            error.attempts = attempt
            wait = policy.wait_for(attempt, error, rng) if error.retryable else None
            if wait is None or attempt == policy.max_attempts:
                raise
            sleep(wait)
    raise ValueError("max_attempts must be at least 1")


class RetryingClient:
    """Wraps any LLMClient and is one itself."""

    def __init__(self, inner: LLMClient, policy: RetryPolicy = RetryPolicy(),
                 sleep: Callable[[float], None] = time.sleep, rng: random.Random | None = None) -> None:
        self.inner, self.policy, self.sleep, self.rng = inner, policy, sleep, rng

    def complete(self, request: LLMRequest) -> LLMResponse:
        return call_with_retry(lambda: self.inner.complete(request), self.policy, self.sleep, self.rng)

import random

from evidencedesk.errors import LLMError, RetryPolicy, call_with_retry, from_status

print(from_status(429, {"Retry-After": "3"}).kind, from_status(429, {}, "credit_balance_exhausted").kind)
tries, waits = [], []


def flaky():
    tries.append(1)
    if len(tries) < 3:
        raise from_status(503)
    return "ok"


print(call_with_retry(flaky, RetryPolicy(), waits.append, random.Random(7)), len(tries), [round(w, 2) for w in waits])


def always_down():
    raise from_status(500)


more = []
try:
    call_with_retry(always_down, RetryPolicy(max_attempts=3), more.append, random.Random(7))
except LLMError as error:
    print("gave up:", error, "after", error.attempts, "attempts;", len(more), "waits")

rate_limited quota_exhausted
ok 3 [0.42, 0.92]
gave up: server: HTTP 500 after 3 attempts; 2 waits

# file: tests/test_config.py
import re
from pathlib import Path

import pytest
from pydantic import ValidationError

from evidencedesk.config import load_settings

ROOT = Path(__file__).resolve().parent.parent
KEY_PATTERN = re.compile(r"sk-[A-Za-z0-9_-]{16,}|AIza[0-9A-Za-z_-]{20,}|-----BEGIN [A-Z ]*PRIVATE KEY")
REAL = {"EVIDENCEDESK_PROVIDER": "anthropic", "EVIDENCEDESK_MODEL": "m1"}


def fields(env):
    with pytest.raises(ValidationError) as info:
        load_settings(env)
    return [".".join(map(str, e["loc"])) for e in info.value.errors()]


def test_empty_environment_means_offline_fake():
    s = load_settings({})
    assert (s.provider, s.model, s.api_key, s.max_retries) == ("fake", "fake-model", None, 2)


def test_the_key_comes_from_env_or_file_and_never_appears_in_repr(tmp_path):
    secret = "sk-" + "a" * 24  # made up
    (tmp_path / "key.txt").write_text(secret, encoding="utf-8")
    for source in ({"EVIDENCEDESK_API_KEY": secret}, {"EVIDENCEDESK_API_KEY_FILE": str(tmp_path / "key.txt")}):
        s = load_settings({**REAL, **source})
        assert secret not in repr(s) + str(s) + s.model_dump_json()
        assert s.api_key.get_secret_value() == secret


def test_bad_settings_are_reported_by_field():
    assert fields({"EVIDENCEDESK_PROVIDER": "nope"}) == ["provider"]
    assert fields({"EVIDENCEDESK_TIMEOUT_S": "0"}) == ["timeout_s"]
    assert fields(REAL) == [""]  # a real provider without a key
    assert fields({"EVIDENCEDESK_PROVIDER": "local"}) == [""]  # a real provider without a model


def test_detector_finds_a_fake_key_and_the_source_tree_has_none():
    assert KEY_PATTERN.search("API_KEY = 'sk-" + "b" * 20 + "'")
    for path in [*ROOT.glob("evidencedesk/*.py"), *ROOT.glob("tests/*.py"), *ROOT.glob("fixtures/*.json")]:
        assert not KEY_PATTERN.search(path.read_text(encoding="utf-8")), path.name

# file: tests/test_errors.py
import pytest

from evidencedesk.errors import LLMError, RetryPolicy, call_with_retry, from_status

NO_JITTER = RetryPolicy(max_attempts=4, jitter=0)


@pytest.mark.parametrize("status, headers, body, kind, retryable, retry_after", [
    (400, {}, "", "invalid_request", False, None),
    (401, {}, "", "auth", False, None),
    (429, {"Retry-After": "3"}, "", "rate_limited", True, 3.0),
    (429, {}, "credit_balance_exhausted", "quota_exhausted", False, None),
    (500, {}, "", "server", True, None),
])
def test_status_mapping(status, headers, body, kind, retryable, retry_after):
    error = from_status(status, headers, body)
    assert (error.kind, error.retryable, error.retry_after) == (kind, retryable, retry_after)


class Flaky:
    def __init__(self, *failures):
        self.failures, self.calls = list(failures), 0

    def __call__(self):
        self.calls += 1
        if self.failures:
            raise self.failures.pop(0)
        return "ok"


def run(fn, policy=NO_JITTER):
    slept = []
    try:
        return call_with_retry(fn, policy, slept.append), slept
    except LLMError as error:
        return error, slept


def test_recovers_and_never_really_sleeps():
    fn = Flaky(from_status(503), from_status(503))
    assert run(fn) == ("ok", [0.5, 1.0]) and fn.calls == 3  # base 0.5, doubled once


def test_attempts_are_bounded():
    fn = Flaky(*[from_status(500)] * 10)
    error, slept = run(fn)
    assert fn.calls == 4 and len(slept) == 3 and error.attempts == 4


def test_permanent_errors_are_not_retried():
    fn = Flaky(from_status(401), from_status(401))
    assert run(fn)[1] == [] and fn.calls == 1


def test_server_wait_wins_but_only_up_to_the_cap():
    assert run(Flaky(from_status(429, {"retry-after": "2"})))[1] == [2.0]
    fn = Flaky(from_status(429, {"retry-after": "60"}))
    assert run(fn)[1] == [] and fn.calls == 1  # 60 s is over max_delay: give up, do not hammer

python -m pytest -q tests/test_config.py tests/test_errors.py

.............                                                            [100%]
13 passed

# file: requirements.txt
# Tested here with Python 3.13; the target is Python 3.12.
pydantic==2.14.1
pytest==9.1.1
# openai==3.29.0  (optional adapters: latest on 2026-10-12, NOT tested here)
# anthropic==1.13.0

# file: fixtures/provider_samples.json
{
  "_note": "Hand-written. Not captured from a real service.",
  "first": {"status": "completed", "output_text": "Returns: 30 days, full refund.",
            "usage": {"input_tokens": 52, "output_tokens": 9}},
  "second": {"stop_reason": "end_turn",
             "content": [{"type": "text", "text": "Returns: 30 days, "}, {"type": "text", "text": "full refund."}],
             "usage": {"input_tokens": 52, "output_tokens": 9}},
  "errors": [
    {"name": "rate limit with a wait", "status": 429, "headers": {"Retry-After": "3"},
     "body": "{\"error\": {\"type\": \"rate_limit_error\"}}", "kind": "rate_limited"},
    {"name": "credit used up", "status": 429, "headers": {},
     "body": "{\"error\": {\"code\": \"credit_balance_exhausted\"}}", "kind": "quota_exhausted"},
    {"name": "overloaded", "status": 529, "headers": {}, "body": "overloaded_error", "kind": "server"}
  ]
}

# file: evidencedesk/summary.py
from __future__ import annotations

from collections.abc import Callable

from evidencedesk.config import Settings
from evidencedesk.errors import LLMError, RetryingClient, RetryPolicy, from_status
from evidencedesk.llm import FakeClient, LLMClient, LLMRequest, LLMResponse, Message, Usage

DEMO_TEXT = ("You can return a new, unridden bike within 30 days of delivery for a full refund. We refund the "
             "price of the bike; the delivery fee is refunded only if the bike arrived damaged.")
DEMO_FIXTURES = {DEMO_TEXT: {"text": "A new, unridden bike can be returned within 30 days for a full refund; "
                                     "the delivery fee is refunded only if it arrived damaged.",
                             "usage": {"input_tokens": 52, "output_tokens": 24}}}


def summarise(client: LLMClient, text: str, max_words: int = 30) -> str:
    """The word limit is enforced here, in code, not only requested in the prompt."""
    text = text.strip()
    if not text:
        raise ValueError("nothing to summarise")
    system = f"Summarise the text in at most {max_words} words. Use only the text."
    response = client.complete(LLMRequest((Message("system", system), Message("user", text))))
    words = response.text.split()
    return " ".join(words[:max_words]) + (" ..." if len(words) > max_words or response.stop_reason == "length" else "")


# Version-sensitive: only these functions know a provider's field names.
def split_system(request: LLMRequest) -> tuple[str, str]:
    def join(role: str) -> str:
        return "\n".join(m.content for m in request.messages if m.role == role)

    return join("system"), join("user")


def from_first(r, model: str) -> LLMResponse:  # r: an object with .output_text, .usage, .status
    stop = "length" if r.status == "incomplete" else "end" if r.status == "completed" else "other"
    return LLMResponse(r.output_text, model, Usage(r.usage.input_tokens, r.usage.output_tokens), stop)


def from_second(r, model: str) -> LLMResponse:  # r: an object with .content blocks, .usage, .stop_reason
    text = "".join(b.text for b in r.content if b.type == "text")
    return LLMResponse(text, model, Usage(r.usage.input_tokens, r.usage.output_tokens),
                       {"end_turn": "end", "max_tokens": "length"}.get(r.stop_reason, "other"))


def guard(call: Callable[[], object], timeout_type: type[Exception]):
    """Run an SDK call and translate its failures."""
    try:
        return call()
    except timeout_type as e:
        raise LLMError("timeout", str(e)) from e
    except Exception as e:
        if getattr(e, "status_code", None) is None:
            raise
        raise from_status(e.status_code, dict(e.response.headers), e.response.text) from e


REGISTRY: dict[str, Callable[[Settings], LLMClient]] = {"fake": lambda s: FakeClient(DEMO_FIXTURES)}


def build_client(settings: Settings) -> LLMClient:
    if settings.provider not in REGISTRY:
        raise ValueError(f"no adapter registered for {settings.provider!r}")
    return RetryingClient(REGISTRY[settings.provider](settings), RetryPolicy(max_attempts=settings.max_retries + 1))

# file: tests/test_summariser.py
import json
from pathlib import Path
from types import SimpleNamespace

import pytest

from evidencedesk.config import load_settings
from evidencedesk.errors import LLMError, from_status
from evidencedesk.llm import FakeClient, LLMClient
from evidencedesk.summary import DEMO_TEXT, build_client, from_first, from_second, guard, summarise

SAMPLES = json.loads((Path(__file__).parent.parent / "fixtures" / "provider_samples.json").read_text(encoding="utf-8"))


def obj(d):  # dictionary -> object with attributes, like an SDK result
    return json.loads(json.dumps(d), object_hook=lambda x: SimpleNamespace(**x))


def test_empty_environment_gives_a_working_offline_summariser():
    client = build_client(load_settings({}))
    assert isinstance(client, LLMClient)
    assert summarise(client, DEMO_TEXT).startswith("A new, unridden bike")


def test_word_limit_is_enforced_in_code():
    long = {"text": "one two three four five six", "usage": {"input_tokens": 1, "output_tokens": 6}}
    assert summarise(FakeClient({"x": long}), "x", max_words=4) == "one two three four ..."


def test_two_provider_shapes_become_one_result():
    results = [from_first(obj(SAMPLES["first"]), "m"), from_second(obj(SAMPLES["second"]), "m")]
    assert results[0] == results[1]  # same text, usage and stop reason from two different shapes
    assert results[0].text == "Returns: 30 days, full refund." and results[0].usage.total_tokens == 61


@pytest.mark.parametrize("case", SAMPLES["errors"], ids=lambda c: c["name"])
def test_recorded_failures_map_to_the_expected_kind(case):
    assert from_status(case["status"], case["headers"], case["body"]).kind == case["kind"]


def test_guard_translates_sdk_failures():
    class Boom(Exception):
        status_code, response = 429, SimpleNamespace(headers={"retry-after": "2"}, text="")

    def fail():
        raise Boom()

    with pytest.raises(LLMError) as info:
        guard(fail, TimeoutError)
    assert (info.value.kind, info.value.retry_after) == ("rate_limited", 2.0)

python -m pytest -q tests/test_summariser.py

.......                                                                  [100%]
7 passed

# not run here: needs the SDK, a key and internet
import openai
from evidencedesk.summary import from_first, guard, split_system

class FirstAdapter:
    def __init__(self, s):  # max_retries=0: our RetryingClient retries
        self.model = s.model
        self.api = openai.OpenAI(api_key=s.api_key.get_secret_value(), timeout=s.timeout_s, max_retries=0)
    def complete(self, request):
        system, user = split_system(request)
        call = lambda: self.api.responses.create(model=self.model, instructions=system, input=user)
        return from_first(guard(call, openai.APITimeoutError), self.model)

# not run here: needs the SDK, a key and internet
import anthropic
from evidencedesk.summary import from_second, guard, split_system

class SecondAdapter:
    def __init__(self, s):
        self.model = s.model
        self.api = anthropic.Anthropic(api_key=s.api_key.get_secret_value(), timeout=s.timeout_s, max_retries=0)
    def complete(self, request):
        system, user = split_system(request)
        ask = dict(model=self.model, max_tokens=request.max_tokens, system=system, messages=[{"role": "user", "content": user}])
        return from_second(guard(lambda: self.api.messages.create(**ask), anthropic.APITimeoutError), self.model)

# file: tests/test_exercises_p02.py
from test_llm import Q_RETURN, ShapeBFake

from evidencedesk.errors import RetryingClient, RetryPolicy, from_status
from evidencedesk.llm import answer_question


class OnceLimited(ShapeBFake):
    todo = [from_status(429, {"Retry-After": "1"})]

    def complete(self, request):
        if self.todo:
            raise self.todo.pop()
        return super().complete(request)


def test_medium_retry_wrapper_around_an_earlier_fake():
    slept = []
    client = RetryingClient(OnceLimited(), RetryPolicy(jitter=0), slept.append)
    assert answer_question(client, Q_RETURN).startswith("You can return") and slept == [1.0]


def public_status(error):  # never pass a provider 401 on: it is our key, not the caller's
    return {"timeout": 504, "rate_limited": 503, "server": 503, "auth": 502, "quota_exhausted": 502}.get(error.kind, 500)


def test_hard_provider_errors_become_our_own_http_statuses():
    assert [public_status(from_status(s)) for s in (401, 429, 400)] == [502, 503, 500]