অধ্যায় 5 · AI কাজের জন্য Python
AI অ্যাপের জন্য অ্যাডভান্সড Python
- পৃষ্ঠা 18 / 20
- 6 মিনিট পড়া
আপনি এখন মডেল ব্যবহার করতে পারেন। আসল AI অ্যাপ্লিকেশনের জন্য আরও কয়েকটা হাতিয়ার লাগে: বড় কোড ঠিক রাখতে টাইপ হিন্ট, কারণ মডেলের আউটপুট চোখ বুজে বিশ্বাস করা যায় না তাই যাচাই (validation), একসাথে অনেক API কল করতে async, উত্তর লেখার সাথে সাথে ব্যবহারকারী যেন দেখতে পায় সেজন্য স্ট্রিমিং, আর চালু অবস্থায় কী ঘটল জানতে লগিং।
টাইপ হিন্ট
টাইপ হিন্ট লিখে রাখে প্রতিটা মান কোন টাইপের হওয়া উচিত। প্রোগ্রাম চলার সময় Python নিজে এগুলো বাধ্যতামূলক করে না, কিন্তু আপনার এডিটর আর mypy বা pyright-এর মতো চেকার এগুলো দিয়ে চালানোর আগেই ভুল ধরে — আর পরের পাঠকের জন্য কোডের ব্যাখ্যাও হয়ে থাকে:
from dataclasses import dataclass
@dataclass
class Document:
title: str
text: str
tags: list[str]
def search(docs: list[Document], word: str, limit: int = 3) -> list[str]:
"""যেসব ডকুমেন্টে `word` আছে তাদের শিরোনাম, সর্বোচ্চ `limit`টা।"""
hits = [d.title for d in docs if word.lower() in d.text.lower()]
return hits[:limit]
def first_tag(doc: Document) -> str | None:
return doc.tags[0] if doc.tags else None
docs = [
Document("Intro", "Python is a language for AI.", ["basics"]),
Document("Pricing", "Plans start at 10 dollars.", []),
]
print(search(docs, "python"))
print(first_tag(docs[1]))['Intro']
Nonelist[str]হলো স্ট্রিংয়ের লিস্ট;dict[str, int]স্ট্রিং থেকে int-এ ম্যাপ করে।str | Noneমানে "একটা স্ট্রিং, অথবা None" — অনুপস্থিত হতে পারে এমন যেকোনো কিছুর জন্য।- বন্ধনীর পরে
-> list[str]বলে ফাংশন কী ফেরত দেয়।
মডেলের আউটপুট কখনো বিশ্বাস করবেন না: যাচাই করুন
মডেলের কাছে JSON চাইলে সাধারণত চাওয়া ঘরসহ বৈধ JSON পাবেন। সাধারণত মানে সবসময় নয়। তার উত্তরকে ব্যবহারকারীর ইনপুটের মতো ধরুন: পার্স করুন, প্রতিটা ঘর যাচাই করুন, আর অপ্রত্যাশিত কিছু হলে বাতিল করুন:
import json
from dataclasses import dataclass
@dataclass
class ReviewAnalysis:
sentiment: str
score: float
topics: list[str]
ALLOWED = {"positive", "negative", "neutral"}
def parse_analysis(raw: str) -> ReviewAnalysis:
"""মডেলের JSON উত্তরকে যাচাই করা Python অবজেক্টে বদলায়, নইলে ValueError তোলে।"""
data = json.loads(raw)
if data.get("sentiment") not in ALLOWED:
raise ValueError(f"unexpected sentiment: {data.get('sentiment')!r}")
score = float(data["score"])
if not 0 <= score <= 1:
raise ValueError(f"score out of range: {score}")
return ReviewAnalysis(data["sentiment"], score, list(data.get("topics", [])))
good = '{"sentiment": "negative", "score": 0.12, "topics": ["delivery", "packaging"]}'
print(parse_analysis(good))
for bad in ['{"sentiment": "angry", "score": 0.3}', '{"sentiment": "positive", "score": 7}', 'Sure! Here is the JSON']:
try:
parse_analysis(bad)
except (ValueError, KeyError) as error:
print("Rejected:", type(error).__name__, "-", error)ReviewAnalysis(sentiment='negative', score=0.12, topics=['delivery', 'packaging'])
Rejected: ValueError - unexpected sentiment: 'angry'
Rejected: ValueError - score out of range: 7.0
Rejected: JSONDecodeError - Expecting value: line 1 column 1 (char 0)ভাঙা JSON-ও একই except-এ ধরা পড়ে, কারণ json.JSONDecodeError এক ধরনের ValueError।
প্রোভাইডাররা স্ট্রাকচার্ড আউটপুট-ও দেয় (আপনি একটা JSON স্কিমা দেন আর মডেল সেটা মানতে বাধ্য থাকে), আর Pydantic-এর মতো লাইব্রেরি এই যাচাইকে একটা ক্লাস সংজ্ঞায় পরিণত করে। এগুলো ব্যর্থতা কমায়, কিন্তু গুরুত্বপূর্ণ মান যাচাইয়ের দায়িত্ব সরায় না — সুগঠিত উত্তরও ভুল হতে পারে।
async: একসাথে অনেক রিকোয়েস্ট
একটা API কলের প্রায় পুরো সময়টাই যায় নেটওয়ার্কের জন্য অপেক্ষায়। তিনটা কলের জন্য একটার পর একটা অপেক্ষা করলে, তিনটার জন্য একসাথে অপেক্ষার চেয়ে তিনগুণ সময় লাগে। async/await একটা প্রোগ্রামকে একসাথে অনেক কিছুর জন্য অপেক্ষা করতে দেয়। এখানে নেটওয়ার্কের বদলে asyncio.sleep:
import asyncio
import time
async def fake_api_call(question: str) -> str:
await asyncio.sleep(1) # নেটওয়ার্কের অপেক্ষার বদলে
return f"answer to {question!r}"
async def main():
questions = ["What is NumPy?", "What is pandas?", "What is an API?"]
start = time.perf_counter()
for q in questions: # একটার পর একটা
await fake_api_call(q)
print(f"One by one: {time.perf_counter() - start:.0f} s")
start = time.perf_counter()
answers = await asyncio.gather(*(fake_api_call(q) for q in questions)) # সব একসাথে
print(f"Together: {time.perf_counter() - start:.0f} s")
print(answers[0])
asyncio.run(main())One by one: 3 s
Together: 1 s
answer to 'What is NumPy?'async defএকটা কোরুটিন সংজ্ঞায়িত করে, এমন ফাংশন যা অপেক্ষার সময় থেমে থাকতে পারে।awaitফলাফল তৈরি না হওয়া পর্যন্ত থামে, এর মধ্যে অন্য কাজ চলতে দেয়।asyncio.gather()কয়েকটা কোরুটিন একসাথে চালায় আর ফলাফলগুলো ক্রম অনুযায়ী জড়ো করে।
আসল SDK দিয়েও দেখতে একই — async ক্লায়েন্ট ব্যবহার করুন আর একসাথে কয়টা রিকোয়েস্ট চলবে তার সীমা দিন, যাতে রেট লিমিটে না পড়েন:
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI()
limit = asyncio.Semaphore(5) # একসাথে সর্বোচ্চ ৫টা রিকোয়েস্ট চলবে
async def summarise(text: str) -> str:
async with limit:
response = await client.responses.create(
model="gpt-5.5", input=f"Summarise in five words: {text}"
)
return response.output_text
async def main():
reviews = ["The delivery was late but the product is great.",
"Terrible battery life, would not buy again.",
"Exactly as described, very happy with it."]
summaries = await asyncio.gather(*(summarise(r) for r in reviews))
for s in summaries:
print("-", s)
asyncio.run(main())উদাহরণ আউটপুট:
- Late delivery, great product.
- Poor battery, would not rebuy.
- Exactly as described, satisfied.স্ট্রিমিং: লেখার সাথে সাথে উত্তর দেখানো
লম্বা উত্তর শেষ হতে অনেক সেকেন্ড লাগতে পারে। stream=True দিলে API তৈরি হওয়ামাত্র ছোট ছোট টুকরো পাঠায়, আর আপনি প্রতিটা সাথে সাথে প্রিন্ট করেন — চ্যাট অ্যাপ এভাবেই "টাইপ" করে। পাতা ৮-এর জেনারেটরের ধারণাই এটা:
from openai import OpenAI
client = OpenAI()
stream = client.responses.create(
model="gpt-5.5",
input="Give me one tip for learning Python.",
stream=True,
)
for event in stream:
if event.type == "response.output_text.delta":
print(event.delta, end="", flush=True) # প্রতিটা টুকরো আসামাত্র দেখাও
print()উদাহরণ আউটপুট (কয়েকটা শব্দ করে করে আসে):
Write a little code every day, and read the error messages carefully.print-এর বদলে লগিং
শেখার সময় print ঠিক আছে। আসল সার্ভিসে logging মডিউল ব্যবহার করুন: প্রতিটা মেসেজের একটা স্তর থাকে (DEBUG, INFO, WARNING, ERROR), কোড না বদলেই কোন স্তর দেখাবেন বাছতে পারেন, আর লগ ফাইলে বা মনিটরিং টুলে পাঠানো যায়:
import logging
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
log = logging.getLogger("review-bot")
def classify(text: str) -> str:
log.debug("classifying %d characters", len(text)) # INFO স্তরে লুকানো
label = "negative" if "bad" in text.lower() else "positive"
log.info("classified as %s", label)
if len(text) < 5:
log.warning("very short input: %r", text)
return label
classify("Bad")
classify("A wonderful phone")INFO review-bot: classified as negative
WARNING review-bot: very short input: 'Bad'
INFO review-bot: classified as positiveযা ডিবাগে সাহায্য করে তা লগ করুন — মডেলের নাম, টোকেনের সংখ্যা, সময়, এরর — কিন্তু API key বা ব্যবহারকারীর ব্যক্তিগত তথ্য কখনো লগ করবেন না।
নিজে চেষ্টা করুন
- এই টিউটোরিয়ালে আগে লেখা তিনটা ফাংশনে টাইপ হিন্ট যোগ করুন।
parse_analysisএমনভাবে বাড়ান, যাতেtopicsসর্বোচ্চ ৫টা স্ট্রিংয়ের লিস্ট হতেই হবে।- async উদাহরণটা বদলে ১০টা প্রশ্ন প্রক্রিয়া করুন, একসাথে সর্বোচ্চ ৩টা চলবে, আর সময় মাপুন।