প্রোভাইডার-নিরপেক্ষ LLM ক্লায়েন্ট

প্রতিটা LLM ফিচারকে নেটওয়ার্কের ওপর দিয়ে কোনো প্রোভাইডারের (provider) সঙ্গে কথা বলতে হয়। সেই কথোপকথন সরাসরি বিজনেস কোডে লিখে ফেললে তিনটা সমস্যা হয়: API কি (key) আর ইন্টারনেট ছাড়া টেস্ট চালানো যায় না, SDK বদলালে প্রতিটা ফাইলে প্রভাব পড়ে, আর অন্য প্রোভাইডারে যাওয়া মানে গোটা কোড নতুন করে লেখা। এই টিউটোরিয়ালে আপনি কাল্পনিক কোম্পানি Kestrel Cycles-এর জন্য EvidenceDesk নামে একটা ছোট অ্যাসিস্ট্যান্ট বানাবেন। এই পাতায় তার ভিত্তি তৈরি হবে: একটা ছোট, প্রোভাইডার-নিরপেক্ষ ক্লায়েন্ট।

HTTP, JSON, ক্লাস, pytest আর Pydantic আপনি Level 0 থেকেই জানেন (AI-এর জন্য Git, API ও ব্যাকএন্ড: বিগিনার, AI-এর জন্য Python: শূন্য থেকে প্রথম AI অ্যাপ)। এখানে সেই জ্ঞান দিয়ে আপনার অ্যাপ্লিকেশন আর যেকোনো প্রোভাইডারের মাঝখানে একটা পরিষ্কার সীমানা টানবেন।

যা শিখবেন

• একটা রিকোয়েস্ট কোন পথে চলে: অ্যাপ্লিকেশনের ইনপুট, অ্যাডাপ্টার, প্রোভাইডারের রিকোয়েস্ট, রেসপন্স আর এরর পথ।
• বিজনেস রুল কীভাবে প্রোভাইডারের কোড থেকে আলাদা রাখবেন, এবং typing.Protocol আর ফ্রোজেন ডেটাক্লাস দিয়ে ছোট একটা LLMClient চুক্তি কীভাবে বানাবেন।
• রেকর্ড করা রেসপন্স ফেরত দেয় এমন নির্ধারিত (deterministic) FakeClient কীভাবে লিখবেন, যাতে প্রতিটা টেস্ট অফলাইনে চলে।
• মডেল API-র HTTP রিকোয়েস্ট ও রেসপন্সে কী কী থাকে, এবং কয়েকটা প্রোভাইডারের (আর একটা লোকাল রানটাইমের) অ্যাডাপ্টার সেগুলোকে আপনার চুক্তিতে কীভাবে বসায়।
• চুক্তিটা কীভাবে প্রমাণ করবেন: ওয়্যার ফরম্যাট আলাদা এমন দুটো ক্লায়েন্টের বিপরীতে একই টেস্ট পাস করে।

এই টিউটোরিয়াল কীভাবে ব্যবহার করবেন

২২টা পাতা

টিউটোরিয়ালে ১০টা পার্ট আছে, প্রতিটায় দুটো করে পাতা; তারপর একটা প্র্যাকটিস ল্যাব আর একটা ক্যাপস্টোন।

সেটআপ

আপনার লাগবে Python 3.12, একটা ভার্চুয়াল এনভায়রনমেন্ট, pytest আর pydantic (পরের পাতা থেকে ব্যবহার হবে)। ইন্টারনেট লাগবে শুধু একবার, pip-এর জন্য। প্রজেক্ট ফোল্ডার তৈরি করুন:

# not run here: installs packages (needs internet once)
python3.12 -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
python -m pip install "pytest>=9.0.3" pydantic
mkdir -p evidencedesk fixtures tests
touch evidencedesk/__init__.py

অফলাইনের নিশ্চয়তা

প্রতিটা ল্যাব চলে একটা ফেক মডেল, রেকর্ড করা ফিক্সচার আর লোকাল ডেটা দিয়ে। API কি (key), মডেল ডাউনলোড, GPU বা ইন্টারনেট না থাকলেও আপনি কখনোই আটকে যাবেন না। আসল প্রোভাইডার আসে শুধু ঐচ্ছিক অ্যাডাপ্টার হিসেবে: তাদের কোডে # not run here লেখা থাকে, কোনো আউটপুট থাকে না, আর তাদের তথ্য থাকে ২০২৬-১০-১২ তারিখের প্রোভাইডার নোটে। প্রোভাইডাররা প্রায়ই বদলায়, তাই প্রতিটা নোটের শেষে বর্তমান ডকুমেন্টেশন দেখতে বলা হয়।

প্রজেক্ট কীভাবে বাড়বে

আপনি একটাই প্যাকেজ বানাবেন, evidencedesk/, প্রতি পাতায় একটা ছোট মডিউল। এই পাতার পর ফোল্ডারটা এমন দেখাবে, আর পরের প্রতিটা পাতা এর পাশে নতুন ফাইল যোগ করবে:

evidencedesk/
  evidencedesk/__init__.py    (empty)
  evidencedesk/llm.py         types, LLMClient, FakeClient     (this page)
  fixtures/responses.json     recorded responses               (this page)
  tests/test_llm.py           contract tests                   (this page)
  corpus/                     13 synthetic Kestrel Cycles help-centre documents (used from Part 2 on)

পরের পাতাগুলোতে যোগ হবে কনফিগারেশন, কনটেক্সট সাজানো, স্ট্রিমিং, এমবেডিং, ইনডেক্স, RAG, একটা টুল, ইভ্যাল, খরচ নিয়ন্ত্রণ, নিরাপত্তা টেস্ট আর প্যাকেজিং। কর্পাসটা কৃত্রিম (synthetic): কোনো আসল কোম্পানি বা গ্রাহকের ডেটা নেই।

ফেক ক্লায়েন্ট নিয়ে একটা সৎ কথা

ফেক ক্লায়েন্ট একটা স্ট্যান্ড-ইন (বদলি), যাতে টেস্ট বারবার একই ফল দেয়। সে এমন লেখা ফেরত দেয় যা একজন মানুষ লিখে রেখেছেন। আসল মডেল কম অনুমানযোগ্য, আর ঠিক সেই কারণেই টেস্ট করতে হয়। ফেক আউটপুট দেখায় আপনার কোড একটা রেসপন্স ঠিকভাবে সামলাচ্ছে কি না; আসল মডেল কী উত্তর দিত, তা কখনোই দেখায় না।

রিকোয়েস্টের জীবনচক্র আঁকুন

Kestrel Cycles-এর একজন গ্রাহকের প্রশ্ন সিস্টেমের ভেতর দিয়ে অনুসরণ করুন। চারটা ধাপ গুরুত্বপূর্ণ, আর ব্যর্থতার জন্য আলাদা একটা পথ আছে:

user question
    |
    v
APPLICATION   validate input, choose instructions, check the result
    |  LLMRequest   (our type)
    v
ADAPTER       map our request to the provider's request, make the call, map the answer back
    |  provider request: HTTP POST with headers and a JSON body
    v
PROVIDER      runs the model
    |  provider response: status code, headers, JSON body, usage numbers
    v
ADAPTER   --- success --->  LLMResponse (our type)  ---> APPLICATION ---> user
    |
    +------- failure ---->  an error the application can understand (next page)

অ্যাপ্লিকেশনের ইনপুট হলো যা আপনার নিজের কোড জানে। অ্যাডাপ্টার একমাত্র কোড যে প্রোভাইডারকে চেনে। এরর পথ সফল পথের মতোই গুরুত্বপূর্ণ: নেটওয়ার্ক টাইমআউট হয়, কোটা ফুরায়, কি-র (key) মেয়াদ শেষ হয়ে যায়। কোন ব্যর্থতায় রিট্রাই করবেন, তা শেখাবে পরের পাতা, অ্যাডাপ্টার, কনফিগারেশন ও এরর হ্যান্ডলিং।

অ্যাপ্লিকেশন লজিক আর প্রোভাইডার কোড আলাদা করুন

প্রতিটা লাইনকে জিজ্ঞেস করুন: "প্রোভাইডার বদলালে কি এই লাইন বদলাত?" উত্তর হ্যাঁ হলে লাইনটা অ্যাডাপ্টারে যাবে। না হলে অ্যাপ্লিকেশনে থাকবে।

সীমানা যাচাইয়ের একটা সহজ পরীক্ষা: প্রতিটা প্রোভাইডার ইমপোর্ট মুছে দিলেও যদি আপনার বিজনেস-রুলের সব টেস্ট পাস করে, তাহলে লাইনটা ঠিক জায়গায় আছে। নিচের টেস্টগুলো ফিচারটাকে কোনো প্রোভাইডার ইনস্টল ছাড়াই চালায়।

একটা ন্যূনতম LLMClient ইন্টারফেস ডিজাইন করুন

চুক্তিটা ইচ্ছা করেই ছোট: একটাই মেথড, complete(request) -> LLMResponse। এতে চারটা ফ্রোজেন ডেটাক্লাস আছে: Message (রোল আর টেক্সট), LLMRequest (মেসেজ, মডেলের নাম, টোকেন সীমা), Usage (ইনপুট ও আউটপুট টোকেনের সংখ্যা) এবং LLMResponse (টেক্সট, মডেল, usage, স্টপ রিজন বা স্টপ রিজন)। LLMClient একটা typing.Protocol, তাই অ্যাডাপ্টারের শুধু মিলে যাওয়া একটা complete মেথড লাগে; কারও কাছ থেকে ইনহেরিট করতে হয় না। frozen=True মানগুলোকে অপরিবর্তনীয় ও তুলনাযোগ্য করে, ফলে টেস্ট সহজ হয়।

ফাইলে FakeClient (পরের সেকশনে) আর একটা ছোট ফিচার answer_question-ও আছে, যেটা অ্যাপ্লিকেশন লজিক: সে ফাঁকা ইনপুট বাতিল করে, রিকোয়েস্ট বানায় আর কাটা পড়া উত্তরে চিহ্ন বসায়। কোনো প্রোভাইডার সে ইমপোর্ট করে না। ফাইলগুলো তৈরি করুন:

# file: evidencedesk/llm.py
"""Provider-neutral LLM contract: types, the client protocol, a deterministic fake, one tiny feature."""
from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol, runtime_checkable

FIXTURES = Path(__file__).resolve().parent.parent / "fixtures" / "responses.json"
SYSTEM = "Answer only from the Kestrel Cycles help centre. Be brief."


@dataclass(frozen=True)
class Message:
    role: str  # "system", "user" or "assistant"
    content: str


@dataclass(frozen=True)
class LLMRequest:
    messages: tuple[Message, ...]
    model: str = "fake-model"
    max_tokens: int = 256


@dataclass(frozen=True)
class Usage:
    input_tokens: int = 0
    output_tokens: int = 0

    @property
    def total_tokens(self) -> int:
        return self.input_tokens + self.output_tokens


@dataclass(frozen=True)
class LLMResponse:
    text: str
    model: str
    usage: Usage = Usage()
    stop_reason: str = "end"  # our own words: "end", "length" or "other"


@runtime_checkable
class LLMClient(Protocol):
    def complete(self, request: LLMRequest) -> LLMResponse: ...


class FixtureMissing(LookupError):
    """The fake has no recorded response for this input."""


def last_user_text(request: LLMRequest) -> str:
    return next(m.content for m in reversed(request.messages) if m.role == "user").strip()


class FakeClient:
    """Returns recorded responses selected by the last user message. A stand-in, not a model."""

    def __init__(self, fixtures: dict[str, dict] | None = None) -> None:
        self.fixtures = fixtures if fixtures is not None else json.loads(FIXTURES.read_text(encoding="utf-8"))
        self.calls: list[LLMRequest] = []

    def complete(self, request: LLMRequest) -> LLMResponse:
        self.calls.append(request)
        key = last_user_text(request)
        if key not in self.fixtures:
            raise FixtureMissing(f"no recorded response for {key!r}")
        rec = self.fixtures[key]
        usage = Usage(**rec["usage"])
        return LLMResponse(rec["text"], request.model, usage, rec.get("stop_reason", "end"))


def answer_question(client: LLMClient, question: str, system: str = SYSTEM) -> str:
    """Application logic: it knows our rules, not any provider."""
    question = question.strip()
    if not question:
        raise ValueError("empty question")
    request = LLMRequest((Message("system", system), Message("user", question)))
    response = client.complete(request)
    return response.text.strip() + (" [cut off]" if response.stop_reason == "length" else "")

লক্ষ করুন, ইচ্ছা করে কী বাদ দেওয়া হয়েছে। temperature নেই: সেটা সমর্থিত কি না তা মডেলের ওপর নির্ভর করে, আর আসল মডেলে temperature 0 দিলেও হুবহু একই আউটপুটের নিশ্চয়তা নেই। স্ট্রিমিং, টুল আর কাঠামোবদ্ধ আউটপুট আসবে পরের পাতাগুলোতে। stop_reason আমাদের নিজেদের তিনটা শব্দ ব্যবহার করে, ফলে কোনো প্রোভাইডারের শব্দ অ্যাপ্লিকেশনে ঢোকে না। @runtime_checkable দিয়ে isinstance(x, LLMClient) লেখা যায়, কিন্তু তা শুধু দেখে যে complete মেথড আছে, তার সিগনেচার ঠিক কি না দেখে না।

প্রোভাইডার নোট (২০২৬-১০-১২ অনুযায়ী): অ্যাডাপ্টারকে কখনো আপনার রিকোয়েস্টের কোনো ফিল্ড উপেক্ষা করতে হতে পারে। যেমন Anthropic-এর Python SDK v1 থেকে temperature, top_p আর top_k সরিয়ে দেওয়া হয়েছে, আর নতুন Claude মডেলে ডিফল্ট ছাড়া অন্য মান দিলে HTTP 400 আসে। কোনো ফিল্ড গ্রহণযোগ্য কি না তা মডেলের ওপর নির্ভর করে, তাই বর্তমান ডকুমেন্টেশন দেখুন।

একটা নির্ধারিত ফেক ক্লায়েন্ট বানান

ফেক ক্লায়েন্ট রেকর্ড করা রেসপন্স পড়ে একটা ছোট JSON ফাইল থেকে, যেখানে কি (key) হলো টেস্টের হুবহু ইনপুট (শেষ ব্যবহারকারী-মেসেজ)। লেখা আর সংখ্যাগুলো হাতে লেখা; এগুলো কোনো পরিমাপ নয়। ফিক্সচার ফাইল তৈরি করুন:

# file: fixtures/responses.json
{
  "How long do I have to return a new bike?": {
    "text": "You can return a new, unridden bike within 30 days of delivery for a full refund.",
    "usage": {"input_tokens": 41, "output_tokens": 19},
    "stop_reason": "end"
  },
  "How long does standard delivery take?": {
    "text": "Standard delivery takes 4 to 7 business days. Express delivery takes 2 business days.",
    "usage": {"input_tokens": 38, "output_tokens": 20},
    "stop_reason": "end"
  },
  "How should I store the e-bike battery for a long time?": {
    "text": "For storage longer than 30 days, keep the battery at 40 to 60 percent charge in a dry place and",
    "usage": {"input_tokens": 45, "output_tokens": 20},
    "stop_reason": "length"
  }
}

নিচের ব্লকটা একটা রেকর্ড করা প্রশ্ন দুইবার করে (যাতে দেখা যায় কিছু বদলায় না), তারপর এমন একটা প্রশ্ন করে যার উত্তর কাটা পড়েছিল, তারপর একটা অজানা প্রশ্ন করে:

from evidencedesk.llm import FakeClient, LLMRequest, Message, answer_question

fake = FakeClient()
request = LLMRequest((Message("user", "How long does standard delivery take?"),))
response = fake.complete(request)
print(response.text)
print(response.model, response.usage, response.stop_reason)
print(fake.complete(request) == response)
print(answer_question(fake, "How should I store the e-bike battery for a long time?"))
try:
    answer_question(fake, "Can I pay in instalments?")
except LookupError as error:
    print(type(error).__name__, "-", error)
print(len(fake.calls), "calls recorded")

Standard delivery takes 4 to 7 business days. Express delivery takes 2 business days.
fake-model Usage(input_tokens=38, output_tokens=20) end
True
For storage longer than 30 days, keep the battery at 40 to 60 percent charge in a dry place and [cut off]
FixtureMissing - no recorded response for 'Can I pay in instalments?'
4 calls recorded

চারটা জিনিস লক্ষ করুন। দুইবার জিজ্ঞেস করলেও ফল সমান। length স্টপ রিজনের জন্য [cut off] যোগ করেছে ফিচার, ফেক নয়, তাই আসল কাটা উত্তর ছাড়াই আপনি এই নিয়মটা টেস্ট করতে পারেন। অজানা প্রশ্নে FixtureMissing ওঠে, উত্তর বানিয়ে বলা হয় না, ফলে ফিক্সচার না থাকলে ব্যর্থতাটা জোরালোভাবে চোখে পড়ে। আর calls প্রতিটা রিকোয়েস্ট মনে রাখে, তাই টেস্ট দেখতে পারে অ্যাপ্লিকেশন কী পাঠিয়েছিল।

HTTP রিকোয়েস্ট ও রেসপন্স পরীক্ষা করুন

প্রতিটা SDK-র নিচে আছে একটা HTTP আদান-প্রদান। রিকোয়েস্টে থাকে মেথড ও URL, হেডার (অন্তত Content-Type: application/json, আর একটা ক্রেডেনশিয়াল হেডার যার নাম প্রোভাইডার অনুযায়ী আলাদা) এবং মডেলের নাম ও মেসেজসহ একটা JSON বডি। রেসপন্সে থাকে স্ট্যাটাস কোড, হেডার এবং তৈরি করা টেক্সট, থামার সূচক আর usage মেটাডেটাসহ একটা JSON বডি। নিচের ব্লক urllib.request.Request দিয়ে একটা লোকাল রানটাইমের চ্যাট এন্ডপয়েন্টের রিকোয়েস্ট বানায় কিন্তু কখনো পাঠায় না, তাই যেকোনো জায়গায় চলে:

import json
import urllib.request

from evidencedesk.llm import LLMRequest, Message

ours = LLMRequest((Message("system", "Be brief."), Message("user", "How long does standard delivery take?")), model="<local-model>")
body = {"model": ours.model, "messages": [{"role": m.role, "content": m.content} for m in ours.messages], "stream": False}
http = urllib.request.Request(
    "http://localhost:11434/api/chat",
    data=json.dumps(body).encode("utf-8"),
    headers={"Content-Type": "application/json"},
    method="POST",
)  # built, never sent: no urlopen call
print(http.get_method(), http.full_url)
print(dict(http.header_items()))
print(json.dumps(json.loads(http.data), indent=2))

POST http://localhost:11434/api/chat
{'Content-type': 'application/json'}
{
  "model": "<local-model>",
  "messages": [
    {
      "role": "system",
      "content": "Be brief."
    },
    {
      "role": "user",
      "content": "How long does standard delivery take?"
    }
  ],
  "stream": false
}

লক্ষ করুন, urllib হেডারের নামকে Content-type করে দিয়েছে; HTTP হেডারের নামে বড়-ছোট হাতের অক্ষর ধর্তব্য নয়। বডিটা আমাদের Message তালিকারই ডিকশনারি-রূপ। এই কলের একটা প্রোভাইডার রেসপন্স দেখতে নিচের হাতে লেখা উদাহরণের মতো। এটা লোকাল রানটাইমের ডকুমেন্টে থাকা ফিল্ড অনুযায়ী সাজানো; মানগুলো বানানো, কোনো সার্ভার থেকে কিছু আসেনি:

HTTP/1.1 200 OK
Content-Type: application/json

{"model": "<local-model>", "message": {"role": "assistant", "content": "Standard delivery takes 4 to 7 business days."},
 "done": true, "prompt_eval_count": 41, "eval_count": 19}

অ্যাডাপ্টার এখান থেকে তিনটা জিনিস পড়ে: টেক্সট (message.content), usage (prompt_eval_count ও eval_count) এবং তৈরি শেষ হয়েছে কি না (done)। সে সবার আগে স্ট্যাটাস কোডও দেখে, কারণ স্ট্যাটাস 200 না হলে পড়ার মতো কোনো উত্তর নেই। পরের ব্লক ওপরের ডিকশনারি আর একটা কল্পিত 429 স্ট্যাটাস নিয়ে সেই ম্যাপিং করে:

from evidencedesk.llm import LLMResponse, Usage

raw = {"model": "<local-model>", "message": {"role": "assistant", "content": "Standard delivery takes 4 to 7 business days."},
       "done": True, "prompt_eval_count": 41, "eval_count": 19}


def from_chat_json(raw: dict, status: int = 200) -> LLMResponse:
    if status != 200:
        raise RuntimeError(f"HTTP {status}")
    usage = Usage(raw["prompt_eval_count"], raw["eval_count"])
    return LLMResponse(raw["message"]["content"], raw["model"], usage, "end" if raw["done"] else "other")


response = from_chat_json(raw)
print(response.text)
print(response.usage, response.usage.total_tokens)
try:
    from_chat_json({}, status=429)
except RuntimeError as error:
    print("error path:", error)

Standard delivery takes 4 to 7 business days.
Usage(input_tokens=41, output_tokens=19) 60
error path: HTTP 429

অ্যাডাপ্টারের কাজের গড়নটা লক্ষ করুন: আগে স্ট্যাটাস যাচাই, তারপর ফিল্ড ম্যাপিং, তারপর আমাদের টাইপ। মোটামুটি, 400, 401, 403 ও 404 মানে রিকোয়েস্ট বা অ্যাক্সেস ভুল (রিট্রাই নয়), 429 মানে রেট লিমিট বা ফুরিয়ে যাওয়া কোটা (রিট্রাই শুধু রেট লিমিট হলে), আর 500 ও তার ওপরে মানে সার্ভারের সমস্যা (সীমিত ব্যাকঅফসহ রিট্রাই)। টাইমআউট আলাদা ব্যাপার: কোনো স্ট্যাটাসই পাওয়া যায় না, তাই প্রোভাইডার কাজটা করেছে কি না আপনি জানেন না। কলে সবসময় টাইমআউট দিন (নিচের লোকাল অ্যাডাপ্টারে 60 সেকেন্ড, এটা আমাদের নিজের বেছে নেওয়া মান), একে রিট্রাইযোগ্য ধরুন শুধু সীমিত সংখ্যক চেষ্টা পর্যন্ত, আর সীমা ছাড়া কখনো রিট্রাই করবেন না। আসল শ্রেণিবিন্যাস বানানো হবে পরের পাতায়।

প্রোভাইডার নোট (২০২৬-১০-১২ অনুযায়ী): usage-এর সংখ্যা এক এক জায়গায় থাকে: usage.input_tokens ও output_tokens (OpenAI Responses, Anthropic Messages), candidatesTokenCount-এর মতো নামসহ usageMetadata (Gemini generateContent), total_input_tokens ও অনুরূপ নাম (Gemini Interactions), prompt_eval_count ও eval_count (Ollama REST)। রেট-লিমিট হেডার: OpenAI পাঠায় x-ratelimit-* আর কখনো Retry-After; Anthropic পাঠায় retry-after, anthropic-ratelimit-* ও request-id; Gemini-র রেট-লিমিট পাতায় এমন কোনো হেডারের বর্ণনা নেই। বর্তমান ডকুমেন্টেশন দেখুন।

প্রথম প্রোভাইডার অ্যাডাপ্টার যোগ করুন

অ্যাডাপ্টার হলো complete মেথডসহ একটা ক্লাস, যা দুই দিকেই অনুবাদ করে। তার তিনটা কাজ: LLMRequest-কে প্রোভাইডারের রিকোয়েস্টে রূপান্তর করা, প্রোভাইডারকে ডাকা, আর প্রোভাইডারের উত্তরকে LLMResponse-এ রূপান্তর করা। চতুর্থ কাজ, ব্যর্থতা ম্যাপ করা, আসবে পরের পাতায়। নিচের স্কেচ প্রথম প্রোভাইডারের SDK ব্যবহার করে। অ্যাকাউন্ট ছাড়া এটা চালানো যায় না, তাই কোনো আউটপুট নেই; পড়ে দেখুন এতে বিজনেস লজিক কতটুকু: একদমই নেই।

# not run here: needs the provider SDK, an API key and internet access
from openai import OpenAI

from evidencedesk.llm import LLMRequest, LLMResponse, Usage


class OpenAIAdapter:
    def __init__(self, model: str) -> None:
        self.client, self.model = OpenAI(), model  # reads OPENAI_API_KEY

    def complete(self, request: LLMRequest) -> LLMResponse:
        system = "\n".join(m.content for m in request.messages if m.role == "system")
        user = "\n".join(m.content for m in request.messages if m.role == "user")
        r = self.client.responses.create(model=self.model, instructions=system, input=user)
        usage = Usage(r.usage.input_tokens, r.usage.output_tokens)
        return LLMResponse(r.output_text, self.model, usage, "end" if r.status == "completed" else "other")

লক্ষ করুন, সিস্টেম টেক্সট যায় instructions-এ, উত্তর আসে output_text থেকে, আর usage-এর ফিল্ডগুলো আমাদের Usage-এ নতুন নামে বসে। স্ট্যাটাস আমাদের দুটো শব্দে ম্যাপ হয়। আমাদের max_tokens এই স্কেচে পাঠানো হয়নি; সেটা পাঠানো হবে কি না, হলে কীভাবে, তা বর্তমান ডকুমেন্টেশন দেখে ঠিক করতে হবে। আসল API কি দিয়ে চালালে উত্তরের টেক্সট আসে মডেল থেকে, তাই তার হুবহু শব্দের ওপর অ্যাসার্ট করবেন না। এখানে কোনো আউটপুট ছাপা হয়নি, কারণ কোনো আউটপুট তৈরি হয়নি।

প্রোভাইডার নোট (২০২৬-১০-১২ অনুযায়ী): OpenAI-এর Python প্যাকেজ openai (pip install openai; ওই তারিখে PyPI-তে সর্বশেষ ছিল 3.29.0)। তাদের SDK পাতা Responses API-কে প্রাথমিক API বলে; পুরোনো Chat Completions API এখনো সমর্থিত। কলটা client.responses.create(model=..., input=...), সিস্টেম-ধরনের নির্দেশনা instructions=-এ, টেক্সট response.output_text-এ, আর response.status-এর মান যেমন completed ও incomplete। ডকুমেন্টে উদাহরণ হিসেবে মডেল আইডি gpt-6-astra দেখানো হয়েছিল; এটা শুধু উদাহরণ, প্রোভাইডারের মডেল-পাতা থেকে বর্তমান আইডি কপি করুন। VERIFY — checked 2026-10-12। বর্তমান ডকুমেন্টেশন দেখুন।

দ্বিতীয় প্রোভাইডার অ্যাডাপ্টার যোগ করুন

দ্বিতীয় প্রোভাইডার একই তথ্য অন্যভাবে সাজায়, আর অ্যাডাপ্টারের কাজই সেটা সামলানো। এখানে সিস্টেম টেক্সট মেসেজ নয়, আলাদা একটা ফিল্ড; উদাহরণগুলোতে সবসময় টোকেন সীমা দেওয়া থাকে; উত্তর আসে কনটেন্ট ব্লকের তালিকা হিসেবে; আর স্টপ রিজনের নিজস্ব শব্দ আছে। অ্যাপ্লিকেশন কোড আর আপনার টেস্ট বদলায় না:

# not run here: needs the provider SDK, an API key and internet access
import anthropic

from evidencedesk.llm import LLMRequest, LLMResponse, Usage


class AnthropicAdapter:
    def __init__(self, model: str) -> None:
        self.client, self.model = anthropic.Anthropic(), model  # reads ANTHROPIC_API_KEY

    def complete(self, request: LLMRequest) -> LLMResponse:
        system = "\n".join(m.content for m in request.messages if m.role == "system")
        chat = [{"role": m.role, "content": m.content} for m in request.messages if m.role != "system"]
        m = self.client.messages.create(model=self.model, max_tokens=request.max_tokens, system=system, messages=chat)
        text = "".join(b.text for b in m.content if b.type == "text")
        return LLMResponse(text, self.model, Usage(m.usage.input_tokens, m.usage.output_tokens),
                           {"end_turn": "end", "max_tokens": "length"}.get(m.stop_reason, "other"))

লক্ষ করুন, দুটো অনুবাদ প্রথম অ্যাডাপ্টারে লাগেনি: তালিকা থেকে সিস্টেম মেসেজ সরানো, আর টেক্সট ব্লক জোড়া লাগানো। স্টপ রিজন একটা ছোট ডিকশনারির মধ্য দিয়ে আমাদের শব্দভান্ডারে আসে; অজানা মান হয় other, কখনো এক্সেপশন নয়।

প্রোভাইডার নোট (২০২৬-১০-১২ অনুযায়ী): Anthropic-এর Python প্যাকেজ anthropic (pip install anthropic; ওই তারিখে PyPI-তে সর্বশেষ 1.13.0)। কলটা client.messages.create(model=..., max_tokens=..., messages=[...]); উদাহরণে max_tokens দেখানো হয়, আর সিস্টেম প্রম্পট মেসেজ নয়, শীর্ষস্তরের system ফিল্ড। উত্তর message.content, ব্লকের তালিকা; টেক্সট ব্লকের type হলো text। স্টপ রিজনের মধ্যে আছে end_turn ও max_tokens। ডকুমেন্টে উদাহরণ মডেল আইডি দেখানো হয়েছিল claude-opus-5-5, এটা শুধু উদাহরণ। VERIFY — checked 2026-10-12। বর্তমান ডকুমেন্টেশন দেখুন।

Google-এর SDK ও লোকাল রানটাইমের তুলনা

আরও দুটো অ্যাডাপ্টার ছবিটা সম্পূর্ণ করে। Google-এরটা Google Gen AI SDK ব্যবহার করে। লোকালটা কোনো SDK-ই নেয় না; ওপরে বানানো HTTP রিকোয়েস্টটাই আপনার নিজের মেশিনে চলা একটা রানটাইমে পাঠায়, যার ডকুমেন্টেড উদাহরণে API কি বা ইন্টারনেট লাগে না:

# not run here: needs the provider SDK, an API key and internet access
from google import genai

from evidencedesk.llm import LLMRequest, LLMResponse


class GeminiAdapter:
    def __init__(self, model: str) -> None:
        self.client, self.model = genai.Client(), model  # reads GEMINI_API_KEY

    def complete(self, request: LLMRequest) -> LLMResponse:
        system = "\n".join(m.content for m in request.messages if m.role == "system")
        user = "\n".join(m.content for m in request.messages if m.role == "user")
        i = self.client.interactions.create(model=self.model, system_instruction=system, input=user)
        return LLMResponse(i.output_text, self.model)  # usage: attribute path not confirmed, check the docs

লক্ষ করুন, usage ম্যাপিং ইচ্ছা করেই বাদ গেছে: ঠিক কোন অ্যাট্রিবিউট পাথ, তা নিশ্চিত করা হয়নি, তাই অনুমান না করে স্কেচটা ডিফল্ট Usage() (শূন্য) ফেরত দেয় এবং কমেন্টে তা বলে। আসল অ্যাডাপ্টার এমন ফাঁক লুকাবে না।

# not run here: needs a local runtime listening on localhost:11434 and a pulled model
import json, urllib.request

from evidencedesk.llm import LLMRequest, LLMResponse, Usage


class LocalAdapter:
    def __init__(self, model: str, url: str = "http://localhost:11434/api/chat") -> None:
        self.model, self.url = model, url

    def complete(self, request: LLMRequest) -> LLMResponse:
        messages = [{"role": m.role, "content": m.content} for m in request.messages]
        body = json.dumps({"model": self.model, "messages": messages, "stream": False}).encode("utf-8")
        http = urllib.request.Request(self.url, body, {"Content-Type": "application/json"})
        raw = json.load(urllib.request.urlopen(http, timeout=60))  # the 60 s is our choice, not a documented default
        usage = Usage(raw["prompt_eval_count"], raw["eval_count"])
        return LLMResponse(raw["message"]["content"], self.model, usage, "end" if raw["done"] else "other")

লক্ষ করুন, এই অ্যাডাপ্টার মানে আগের রিকোয়েস্ট ব্লক আর আগের ম্যাপিং ব্লক একসঙ্গে: আগে মেকানিজম, পরে লাইব্রেরি।

প্রোভাইডার নোট (২০২৬-১০-১২ অনুযায়ী): Gemini-র দুটো API পথ আছে। Google-এর ডকুমেন্টেশন Interactions API দিয়ে শুরু করে (client.interactions.create(model=..., input=...), টেক্সট interaction.output_text-এ, সিস্টেম টেক্সট system_instruction=-এ) এবং নতুন প্রজেক্টের জন্য একে সুপারিশকৃত বলে। পুরোনো generateContent পথ (client.models.generate_content) "legacy" বলা হলেও পুরোপুরি সমর্থিত, আর SDK README এখনো সেটাই আগে দেখায়। প্যাকেজ google-genai (ওই তারিখে সর্বশেষ 2.29.0)। ডকুমেন্টে উদাহরণ মডেল আইডি: gemini-3.8-flash। দুই পথেই usage-এর Python অ্যাট্রিবিউটের নাম নিশ্চিত করা যায়নি। লোকাল রানটাইম: Ollama একটা আলাদা অ্যাপ্লিকেশন; এর API শোনে http://localhost:11434-এ; model, messages ও stream দিয়ে POST /api/chat করলে ফেরে message.content, done, prompt_eval_count ও eval_count; Python প্যাকেজ ollama দেয় chat(...)। আগে মডেলটা নাম ধরে pull করতে হয় (ডকুমেন্টের উদাহরণ: gemma4:e2b)। Python ক্লায়েন্টের ডিফল্ট টাইমআউট ও রিট্রাই নথিভুক্ত নয়, তাই ধরে নিন কিছুই নেই। VERIFY — checked 2026-10-12। বর্তমান ডকুমেন্টেশন দেখুন।

প্রোভাইডার যেটাই হোক, অ্যাডাপ্টারকে খুঁজে বের করতে হবে টেক্সট (কখনো কয়েক অংশে), টোকেনের সংখ্যা (যদি জানানো হয়), জেনারেশন কেন থেমেছে (end, length বা other-এ রূপান্তরিত), আর ব্যর্থতা (আমাদের কোনো একটা এররে রূপান্তরিত, পরের পাতায়)। ফিচারও সমান নয় (স্ট্রিমিং, কাঠামোবদ্ধ আউটপুট, টুল, এমবেডিং), তাই সব প্রোভাইডার একই ফিচার দেয় বলে ধরে নেবেন না; পরের পাতায় একটা ক্যাপাবিলিটি ম্যাট্রিক্স যোগ হবে।

অনুশীলনী: ফিচার না বদলে প্রোভাইডার বদলান

আসল অ্যাডাপ্টার এখানে চালানো যায় না, তাই ধারণাটা অফলাইনে প্রমাণ করুন। নিচের ShapeBFake এমন একটা প্রোভাইডারের অনুকরণ করে যার ওয়্যার ফরম্যাট আলাদা: তার রিকোয়েস্ট বডিতে who ও text সহ prompt আইটেম থাকে, আর রেসপন্সে থাকে parts, tokens এবং একটা finish মান (STOP বা CUT)। from_shape_b ফাংশনটা অ্যাডাপ্টারের ভূমিকা নেয়। তারপর একই টেস্টের সেট দুটো ক্লায়েন্টের বিপরীতেই চলে। ফাইলটা তৈরি করে চালান:

# file: tests/test_llm.py
import dataclasses

import pytest

from evidencedesk.llm import (FakeClient, FixtureMissing, LLMClient, LLMRequest, LLMResponse, Message, Usage,
                              answer_question)

Q_RETURN = "How long do I have to return a new bike?"
Q_BATTERY = "How should I store the e-bike battery for a long time?"


def to_shape_b(request: LLMRequest) -> dict:
    """What a different provider's HTTP body could look like."""
    return {"prompt": [{"who": m.role, "text": m.content} for m in request.messages], "limit": request.max_tokens}


def from_shape_b(raw: dict, model: str) -> LLMResponse:
    """The adapter step: provider shape in, our LLMResponse out."""
    text = "".join(part["text"] for part in raw["parts"])
    usage = Usage(raw["tokens"]["in"], raw["tokens"]["out"])
    return LLMResponse(text, model, usage, {"STOP": "end", "CUT": "length"}.get(raw["finish"], "other"))


class ShapeBFake:
    """Imitates a provider with another wire format; the adapter maps it back."""

    def __init__(self) -> None:
        self.inner = FakeClient()

    def complete(self, request: LLMRequest) -> LLMResponse:
        body = to_shape_b(request)  # the provider only ever sees its own format
        key = [item["text"] for item in body["prompt"] if item["who"] == "user"][-1].strip()
        if key not in self.inner.fixtures:
            raise FixtureMissing(key)
        rec = self.inner.fixtures[key]
        cut = len(rec["text"]) // 2
        raw = {"parts": [{"text": rec["text"][:cut]}, {"text": rec["text"][cut:]}],
               "tokens": {"in": rec["usage"]["input_tokens"], "out": rec["usage"]["output_tokens"]},
               "finish": "CUT" if rec.get("stop_reason") == "length" else "STOP"}
        return from_shape_b(raw, request.model)


@pytest.fixture(params=[FakeClient, ShapeBFake], ids=["fake", "shape-b"])
def client(request):
    return request.param()


def test_same_feature_same_answer(client):
    assert isinstance(client, LLMClient)
    assert answer_question(client, Q_RETURN) == "You can return a new, unridden bike within 30 days of delivery for a full refund."


def test_truncation_rule_lives_in_the_feature(client):
    assert answer_question(client, Q_BATTERY).endswith("and [cut off]")


def test_unknown_input_is_an_error_not_a_guess(client):
    with pytest.raises(FixtureMissing):
        answer_question(client, "Can I pay in instalments?")


def test_empty_question_rejected_before_any_call():
    fake = FakeClient()
    with pytest.raises(ValueError):
        answer_question(fake, "   ")
    assert fake.calls == []


def test_fake_is_deterministic():
    request = LLMRequest((Message("user", Q_RETURN),))
    assert FakeClient().complete(request) == FakeClient().complete(request)


def test_usage_and_frozen_types():
    response = FakeClient().complete(LLMRequest((Message("user", Q_RETURN),)))
    assert response.usage.total_tokens == 60
    with pytest.raises(dataclasses.FrozenInstanceError):
        response.text = "changed"

python -m pytest -q tests/test_llm.py

.........                                                                [100%]
9 passed in 0.09s

লক্ষ করুন, প্রতিটা চুক্তি-টেস্ট দুইবার চলেছে, প্রতি ক্লায়েন্টে একবার (fake ও shape-b প্যারামিটার), অথচ answer_question একবারও বদলানো হয়নি। "ফিচার না বদলে প্রোভাইডার বদলানো" মানে এটাই: শুধু যে অবজেক্টটা দেওয়া হচ্ছে সেটা বদলায়।

এই টেস্ট যা দেখাতে পারে না: এরা প্রমাণ করে আপনার কোড চুক্তি মানছে, কিন্তু প্রমাণ করে না যে একটা আসল অ্যাডাপ্টার আসল প্রোভাইডারকে ঠিকভাবে ম্যাপ করছে। আপনার লেখা ফেক আপনার লেখা অ্যাডাপ্টারের সঙ্গে সহমত হবেই। কোনো অ্যাডাপ্টারের ওপর ভরসা করার আগে আসল সার্ভিসের বিপরীতে একবার চালান (ঐচ্ছিক, এবং এই টিউটোরিয়ালের অফলাইন পথের বাইরে), আর একটা ছোট, পরিষ্কার করা রেকর্ড করা রেসপন্স ফিক্সচার হিসেবে রেখে দিন।

কাজের জায়গায় এটা কোথায় দেখা যায়

দল প্রোভাইডার বদলায় খরচ, মান, ডেটা কোথায় থাকবে সংক্রান্ত নিয়ম বা কোনো বিভ্রাটের কারণে। যে দল সংকীর্ণ ইন্টারফেসের বিপরীতে কোড লিখেছে, তারা একটা অ্যাডাপ্টার বদলে টেস্ট আবার চালায়; যারা SDK কল কোডজুড়ে ছড়িয়ে দিয়েছে, তাদের বদলাতে হয় কয়েক ডজন ফাইল। একই সীমানার কারণে কন্টিনিউয়াস ইন্টিগ্রেশন সিক্রেট ছাড়াই চলতে পারে।

সাধারণ ভুল

• বিজনেস কোডে প্রোভাইডার SDK ইমপোর্ট করা। তখন প্রতিটা টেস্টে API কি লাগে আর প্রতিটা ফাইল একটা ভেন্ডরের সঙ্গে বাঁধা পড়ে। ইমপোর্ট অ্যাডাপ্টারের ভেতরে রাখুন।
• প্রোভাইডারের কাঁচা ডিকশনারি ওপরে পাঠিয়ে দেওয়া। তখন যারা এটা ব্যবহার করে, তারা তার গড়নের ওপর নির্ভর করে। অ্যাডাপ্টারের কিনারায় রূপান্তর করে নিন।
• যে ফেক আন্দাজ করে। অজানা ইনপুটে ডিফল্ট ফেরত দিলে ফিক্সচার না থাকাটা ঢাকা পড়ে, আর টেস্ট ভুল কারণে পাস করে। এরর তুলুন।
• ফেক আউটপুটকে মডেলের আচরণ ভাবা। ফিক্সচার দেখায় আপনার কোড একটা উত্তর কীভাবে সামলায়, আসল মডেল কী বলে তা নয়; temperature 0 হলেও আসল মডেল এক রান থেকে আরেক রানে আলাদা হতে পারে।

নিজে চেষ্টা করুন

1. সহজ। "Where do you deliver?" প্রশ্নের জন্য একটা ফিক্সচার যোগ করুন (টেক্সটের জন্য corpus/-এর শিপিং ডকুমেন্ট দেখুন) এবং দেখান যে llm.py একটুও না বদলে answer_question সেটা ফেরত দেয়।
2. মাঝারি। AI-এর জন্য Git, API ও ব্যাকএন্ড: বিগিনার থেকে HTTP ঝালিয়ে নিন। একটা urllib.request.Request বানান (পাঠাবেন না), যার JSON বডি to_shape_b(request), এবং অ্যাসার্ট করুন মেথড, কনটেন্ট টাইপ, আর বডি ডিকোড করলে একই ডিকশনারি ফেরে।
3. কঠিন। একটা RecordingClient লিখুন, যেটা যেকোনো LLMClient-কে মুড়ে নেয়, প্রতিটা (রিকোয়েস্ট, রেসপন্স) জোড়া একটা তালিকায় রাখে, এবং নিজেও একটা বৈধ LLMClient। ShapeBFake-কে মুড়ে একে টেস্ট করুন।

উত্তর

একটা টেস্ট ফাইলে তিনটা সমাধানই আছে; সেটা আগে লেখা ফাইল থেকে নাম ইমপোর্ট করে।

# file: tests/test_exercises.py
import json
import urllib.request

from test_llm import Q_RETURN, ShapeBFake, to_shape_b

from evidencedesk.llm import FakeClient, LLMClient, LLMRequest, Message, answer_question


def test_easy_new_fixture_without_touching_the_feature():
    question = "Where do you deliver?"
    record = {"text": "We deliver to 12 countries.", "usage": {"input_tokens": 20, "output_tokens": 8}}
    assert answer_question(FakeClient({question: record}), question) == "We deliver to 12 countries."


def test_medium_http_body_round_trip():
    request = LLMRequest((Message("user", "Hi"),), max_tokens=50)
    http = urllib.request.Request("https://llm.example.test/v1/respond", data=json.dumps(to_shape_b(request)).encode(),
                                  headers={"Content-Type": "application/json"}, method="POST")
    assert http.get_header("Content-type") == "application/json"
    assert json.loads(http.data) == {"prompt": [{"who": "user", "text": "Hi"}], "limit": 50}


class RecordingClient:
    def __init__(self, inner: LLMClient) -> None:
        self.inner, self.log = inner, []

    def complete(self, request):
        response = self.inner.complete(request)
        self.log.append((request, response))
        return response


def test_hard_a_wrapper_is_a_client_too():
    client = RecordingClient(ShapeBFake())
    assert isinstance(client, LLMClient)
    answer_question(client, Q_RETURN)
    assert len(client.log) == 1 and client.log[0][1].usage.total_tokens == 60

python -m pytest -q

............                                                             [100%]
12 passed in 0.11s

পাস চেকলিস্ট:

• সব টেস্ট পাস করেছে, আর সহজ কাজের জন্য llm.py বদলানো হয়নি।
• মাঝারি টেস্ট কখনো কানেকশন খোলে না (urlopen নেই)।
• কঠিন কাজের র‍্যাপারে একটা complete(request) মেথড আছে, সে ভেতরের রেসপন্স অপরিবর্তিত ফেরত দেয়, এবং isinstance(client, LLMClient) পাস করে।

সারসংক্ষেপ

• রিকোয়েস্ট চলে অ্যাপ্লিকেশন, অ্যাডাপ্টার, প্রোভাইডার আর আবার ফিরে; এরর পথ ডিজাইনেরই অংশ।
• বিজনেস রুল থাকুক অ্যাপ্লিকেশনে, আর প্রোভাইডার-নির্দিষ্ট সবকিছু অ্যাডাপ্টারে।
• অ্যাডাপ্টার ভেদে সিস্টেম টেক্সট, উত্তর, usage ও স্টপ রিজন আলাদা জায়গায় থাকে, আর প্রোভাইডারদের ফিচারও সমান নয়; প্রতিটা অ্যাডাপ্টার সেই পার্থক্য লুকিয়ে রাখে।
• চুক্তি একটাই মেথড, complete(request) -> LLMResponse, ফ্রোজেন ডেটাক্লাস আর একটা Protocol দিয়ে।
• রেকর্ড করা ফিক্সচারসহ একটা নির্ধারিত ফেক টেস্টকে অফলাইন ও পুনরাবৃত্তিযোগ্য করে, কিন্তু সে একটা স্ট্যান্ড-ইন মাত্র: আসল মডেলের আচরণ সে কখনো দেখায় না।

এরপর: অ্যাডাপ্টার, কনফিগারেশন ও এরর হ্যান্ডলিং। সেখানে আপনি মডেলের নাম আর সেটিংস কোডের বাইরে রাখবেন, সিক্রেট নিরাপদে পড়বেন, আর এররকে রিট্রাই করা যায় এমন আর যায় না এমন, এই দুই ভাগে ভাগ করবেন।