Next.js ছাড়াও — Python দিয়ে কীভাবে একটা Simple AI Agent বানানো যায়

আমাদের আগের পর্বে আমরা যখন এই সিরিজে যা যা বানানো হলো তা জোড়া দিয়ে একটা Complete Mini-Agent প্রজেক্ট রিক্যাপ তৈরি করেছিলাম, তখন ১৪টি পর্বের সবগুলো মডিউল একত্রিত করে Next.js, TypeScript এবং Supabase দিয়ে আমাদের পূর্ণাঙ্গ DevPulse ওয়ার্কস্পেস দাঁড় করিয়েছিলাম। আপনার এজেন্ট যখন সরাসরি ওয়েব ইউআই-এর ভেতরে বসে কাজ করে, তখন Next.js নিঃসন্দেহে সেরা পছন্দ। কিন্তু বাস্তবে এমন অনেক পরিস্থিতি আসে যখন আপনার এজেন্টকে ভারী ডেটা অ্যানালাইসিস করতে হয়, মেশিন লার্নিং বা কম্পিউটার ভিশন মডেল চালাতে হয়, বড় পিডিএফ পার্স করতে হয় কিংবা এমন একটি স্বতন্ত্র ব্যাকএন্ড মাইক্রোসার্ভিস হিসেবে কাজ করতে হয় যাকে ওয়েব অ্যাপ, মোবাইল অ্যাপ এবং টেলিগ্রাম বট সবাই একসাথে কল করবে—তখন আপনার তুরুপের তাস হলো Python।

Next.js বা ভারী কোনো ফ্রেমওয়ার্ক ছাড়া শুধুমাত্র Python দিয়ে একটি পরিচ্ছন্ন ও প্রোডাকশন-গ্রেড AI Agent বানানোর মূল কৌশল হলো—টুলের ইনপুট ভ্যালিডেশনের জন্য Pydantic v2 (BaseModel ও Field), নিরাপদ ও টাইমআউট-যুক্ত এক্সটার্নাল API কলের জন্য httpx, এবং অফিশিয়াল OpenAI বা Anthropic Python SDK ব্যবহার করে একটি নিয়ন্ত্রিত for current_step in range(1, max_steps + 1) রিজনিং লুপ তৈরি করা। এই আর্কিটেকচারটি আমাদের TypeScript এজেন্টের মতোই .model_json_schema() দিয়ে স্বয়ংক্রিয়ভাবে LLM টুল ডেফিনিশন তৈরি করে, রানটাইম ভ্যালিডেশন এরর সামলায় এবং প্রতিটি ধাপের বিস্তারিত ট্রেস (StepTrace) সংরক্ষণ করে।

আজকের এই ২৭তম পর্বের মাধ্যমে আমরা আমাদের ৬২ পর্বের সিরিজের Section C (Backend / অন্য Stack-এ বিস্তৃত হওয়া, পর্ব ২৭–৩৬) শুরু করছি। আজ আমরা Python দিয়ে আমাদের এজেন্টের কোর ইঞ্জিনটি তৈরি করব, যাতে ঠিক পরের পর্বেই (পর্ব ২৮) এটিকে FastAPI দিয়ে একটি পূর্ণাঙ্গ প্রোডাকশন ব্যাকএন্ড API-তে রূপান্তর করা যায়।

Next.js জানার পরও একজন ডেভেলপারের কেন Python দিয়ে Agent বানানো শেখা উচিত?

গত ১৪টি পর্বে আমরা দেখেছি TypeScript এবং Next.js দিয়ে কত চমৎকার এজেন্ট বানানো যায়। তাহলে আবার Python শেখার প্রয়োজন পড়ছে কেন? বাস্তব সফটওয়্যার ইঞ্জিনিয়ারিংয়ে এর ৩টি সুনির্দিষ্ট কারণ রয়েছে:

১. ডেটা সায়েন্স ও AI ইকোসিস্টেমের সরাসরি অ্যাক্সেস: আপনার এজেন্টকে যদি কখনো Pandas দিয়ে বিশাল CSV/Excel ডেটা প্রসেস করতে হয়, কোনো কাস্টম PyTorch বা Scikit-learn মডেল রান করতে হয়, কিংবা লোকাল ডকুমেন্ট OCR ও চ্যাঙ্কিং করতে হয়—তবে সেই লাইব্রেরিগুলোর মূল আবাসস্থল হলো Python। ২. দীর্ঘ সময় ধরে চলা ব্যাকগ্রাউন্ড টাস্ক (Long-Running Jobs): সার্ভারলেস Next.js ফাংশন সাধারণত ১০ থেকে ৬০ সেকেন্ডের রিকোয়েস্টের জন্য তৈরি। কিন্তু একটি Python এজেন্ট সার্ভারে বসে ১০ মিনিট ধরে ৫০টি ফাইল বিশ্লেষণ বা ক্রন জব (Cron Job) অনায়াসে চালাতে পারে। ৩. মাল্টি-প্ল্যাটফর্ম মাইক্রোসার্ভিস আর্কিটেকচার: আপনার কোম্পানির ওয়েবসাইট যদি হয় Next.js বা Laravel-এ, আর মোবাইল অ্যাপ হয় Flutter-এ—তখন দু জায়গায় দুইবার এজেন্ট না লিখে Python দিয়ে একটি কেন্দ্রীয় এজেন্ট সার্ভিস বানিয়ে নিলে যেকোনো প্ল্যাটফর্ম থেকে সেটিকে কল করা যায়।

TypeScript থেকে Python-এ কনসেপ্ট ম্যাপ: কোনটার বদলে কী ব্যবহার করব?

সবচেয়ে আনন্দের বিষয় হলো, আমরা ১২ থেকে ২৬তম পর্বে এজেন্টের যে মূল ইঞ্জিনিয়ারিং নিয়মগুলো শিখেছি, Python-এও তার লজিক হুবহু এক—শুধু টুলগুলোর নাম আলাদা:

• Zod (z.object) → Pydantic v2 (BaseModel ও Field): TypeScript-এ আমরা যেভাবে Zod দিয়ে টুলের আর্গুমেন্ট ভ্যালিডেট করতাম, Python-এ ঠিক সেই কাজটিই করে Pydantic v2। এর .model_json_schema() মেথডটি এক লাইনে আপনার ক্লাসকে LLM-এর উপযোগী JSON Schema-তে রূপান্তর করে দেয়।
• ToolResultEnvelope<T> → Python ToolResultEnvelope(BaseModel): কোনো টুলে এরর হলে প্রসেস ক্র্যাশ না করিয়ে { ok: False, agent_instruction: '...' } অবজেক্ট রিটার্ন করা।
• fetch() + AbortSignal.timeout(5000) → httpx.Client(timeout=5.0): Python-এর পুরনো requests লাইব্রেরির বদলে আধুনিক ও দ্রুতগতির httpx ব্যবহার করা, যাতে ৫ সেকেন্ডের হার্ড টাইমআউট বিল্ট-ইন থাকে।

নিচের ফ্লোচার্টে দেখুন আমাদের Python এজেন্ট কীভাবে একটি নিয়ন্ত্রিত লুপের ভেতরে Pydantic ভ্যালিডেশন এবং টুল এক্সিকিউশন পরিচালনা করে:

mermaid flowchart TD Input([ইউজারের প্রশ্ন / CLI বা API ইনপুট]) --> Loop{Python Agent Loop current_step <= max_steps 5} Loop -->|সিস্টেম প্রম্পট + হিস্ট্রি + Pydantic JSON Schema| LLM[Anthropic Claude 3.5 Sonnet / OpenAI GPT-4o] LLM -->|stop_reason == end_turn| Final[AgentRunResult রিটার্ন final_answer + steps_trace + token_usage] LLM -->|stop_reason == tool_use| Dispatch[টুল রেজিস্ট্রি ডিসপ্যাচার] Dispatch --> Validate[Pydantic v2 model_validate টাইপ ও Allowlist চেক] Validate -->|ValidationError| ErrEnv[সেলফ-কারেকশন এরর এনভেলপ রিটার্ন] Validate -->|সঠিক ইনপুট| Exec[httpx 5s Timeout সহ Python টুল রান] Exec --> Shaper[DTO Shaper ও টেক্সট ট্রিমার] Shaper --> Append[Messages অ্যারেতে tool_result ব্লক যোগ] ErrEnv --> Append Append --> Loop আর্কিটেকচার ফ্লো সামারি: ইউজার ইনপুট → বাউন্ডেড পাইথন লুপ (max_steps = 5) → Pydantic JSON স্কিমাসহ LLM কল → রানটাইম Pydantic ভ্যালিডেশন → httpx টুল এক্সিকিউশন ও DTO শেপিং → মেসেজ হিস্ট্রিতে ফলাফল যোগ → চূড়ান্ত AgentRunResult প্রদান।

ধাপ ১: প্রজেক্ট ফোল্ডার ও Python Virtual Environment সেটআপ

Python প্রজেক্টের প্রথম নিয়ম হলো অপারেটিং সিস্টেমের গ্লোবাল এনভায়রনমেন্ট নষ্ট না করে একটি আলাদা ভার্চুয়াল এনভায়রনমেন্ট (.venv) তৈরি করে নেওয়া। টার্মিনালে নিচের কমান্ডগুলো চালান:

bash mkdir devpulse-python-agent && cd devpulse-python-agent python3 -m venv .venv source .venv/bin/activate pip install anthropic openai pydantic httpx python-dotenv

এবার প্রজেক্টের রুট ফোল্ডারে একটি .env ফাইল তৈরি করে আপনার API Key রাখুন:

`ini

.env

ANTHROPIC_API_KEY=sk-ant-your-key-here OPENAI_API_KEY=sk-proj-your-key-here `

ধাপ ২: Pydantic v2 দিয়ে টুল স্কিমা ও নিরাপদ এক্সিকিউটর তৈরি (agent/tools.py)

চলুন agent/tools.py ফাইলটি তৈরি করি। ১৯তম পর্বে আমরা যেভাবে লাইভ GitHub API কল করার সময় রিপোজিটরি Allowlist, ৫ সেকেন্ডের টাইমআউট এবং ৬০০ ক্যারেক্টারের DTO শেপার বসিয়েছিলাম, এখানেও আমরা Pydantic v2 এবং httpx দিয়ে ঠিক সেই একই প্রোডাকশন স্ট্যান্ডার্ড বজায় রাখব:

`python

agent/tools.py

import time from typing import Any, Callable, Dict, Literal, Optional import httpx from pydantic import BaseModel, Field, ValidationError

AllowedRepo = Literal[ 'vercel/next.js', 'vercel/ai', 'langchain-ai/langchainjs', ]

class SearchDocsInput(BaseModel): query: str = Field( min_length=4, max_length=200, description='Concise technical query for searching internal documentation.', ) top_k: int = Field( default=3, ge=1, le=5, description='Number of documentation chunks to retrieve (1 to 5).', )

class InspectGitHubIssueInput(BaseModel): repository: AllowedRepo = Field( description='Allowlisted GitHub repository in owner/name format.' ) issue_number: int = Field( gt=0, le=999999, description='Positive numeric GitHub issue ID (e.g., 402).', )

class DraftTicketInput(BaseModel): title: str = Field(min_length=10, max_length=120) severity: Literal['low', 'medium', 'high', 'critical'] component: Literal['auth', 'api-gateway', 'database', 'frontend-ui'] reproduction_steps: list[str] = Field(min_length=1)

class ToolResultEnvelope(BaseModel): ok: bool tool_name: str retryable: bool duration_ms: int data: Optional[Dict[str, Any]] = None agent_instruction: Optional[str] = None

def execute_search_docs(args: SearchDocsInput) -> Dict[str, Any]: return { 'query_used': args.query, 'top_k': args.top_k, 'matched_chunks': [ { 'doc_path': 'docs/api/rate-limits.md', 'score': 0.93, 'excerpt': ( 'The API gateway enforces a sliding window of 60 requests ' 'per minute per IP. Bursts during token refresh return HTTP 429 ' 'with a Retry-After header.' ), } ], }

def execute_inspect_github_issue(args: InspectGitHubIssueInput) -> Dict[str, Any]: url = f'https://api.github.com/repos/{args.repository}/issues/{args.issue_number}' headers = { 'Accept': 'application/vnd.github+json', 'User-Agent': 'DevPulse-Python-Agent/1.0', }

with httpx.Client(timeout=5.0) as client:
    response = client.get(url, headers=headers)

if response.status_code == 404:
    raise ValueError(
        f'Issue #{args.issue_number} not found in {args.repository}. Do not retry.'
    )

response.raise_for_status()
raw = response.json()
raw_body = raw.get('body') or 'No description provided.'
clean_excerpt = raw_body[:600] + ('...' if len(raw_body) > 600 else '')

labels = [
    lbl.get('name', '') if isinstance(lbl, dict) else str(lbl)
    for lbl in raw.get('labels', [])[:8]
]

return {
    'repository': args.repository,
    'issue_number': raw.get('number'),
    'title': raw.get('title'),
    'state': raw.get('state'),
    'author': (raw.get('user') or {}).get('login', 'unknown'),
    'labels': [l for l in labels if l],
    'comments_count': raw.get('comments', 0),
    'body_excerpt': clean_excerpt,
}

def execute_draft_ticket(args: DraftTicketInput) -> Dict[str, Any]: return { 'draft_id': 'PY-DRAFT-701', 'status': 'awaiting_human_approval', 'ticket_preview': args.model_dump(), }

TOOL_REGISTRY: Dict[str, tuple[type[BaseModel], Callable[[Any], Dict[str, Any]], str]] = { 'search_knowledge_base': ( SearchDocsInput, execute_search_docs, 'Search internal technical documentation for rate limits, auth rules, and architecture specs.', ), 'inspect_live_github_issue': ( InspectGitHubIssueInput, execute_inspect_github_issue, 'Fetch live status, labels, and truncated body for a numeric issue ID in an allowlisted GitHub repo.', ), 'draft_issue_ticket': ( DraftTicketInput, execute_draft_ticket, 'Create a structured bug ticket draft requiring human approval before database insertion.', ), }

def get_anthropic_tool_schemas() -> list[Dict[str, Any]]: schemas = [] for name, (model_cls, _, description) in TOOL_REGISTRY.items(): schemas.append({ 'name': name, 'description': description, 'input_schema': model_cls.model_json_schema(), }) return schemas

def dispatch_agent_tool(tool_name: str, raw_input: Dict[str, Any]) -> str: start_time = time.perf_counter() if tool_name not in TOOL_REGISTRY: envelope = ToolResultEnvelope( ok=False, tool_name=tool_name, retryable=False, duration_ms=0, agent_instruction=f'Unknown tool {tool_name}. Choose a registered tool.', ) return envelope.model_dump_json()

model_cls, executor_fn, _ = TOOL_REGISTRY[tool_name]

try:
    validated_args = model_cls.model_validate(raw_input)
    data = executor_fn(validated_args)
    duration_ms = int((time.perf_counter() - start_time) * 1000)
    envelope = ToolResultEnvelope(
        ok=True,
        tool_name=tool_name,
        retryable=False,
        duration_ms=duration_ms,
        data=data,
    )
    return envelope.model_dump_json()
except ValidationError as val_err:
    duration_ms = int((time.perf_counter() - start_time) * 1000)
    envelope = ToolResultEnvelope(
        ok=False,
        tool_name=tool_name,
        retryable=True,
        duration_ms=duration_ms,
        agent_instruction=f'Pydantic validation failed: {val_err.errors()}. Fix args and retry.',
    )
    return envelope.model_dump_json()
except Exception as exc:
    duration_ms = int((time.perf_counter() - start_time) * 1000)
    envelope = ToolResultEnvelope(
        ok=False,
        tool_name=tool_name,
        retryable=False,
        duration_ms=duration_ms,
        agent_instruction=f'Tool execution error: {str(exc)}',
    )
    return envelope.model_dump_json()

`

উপরের কোডে get_anthropic_tool_schemas() ফাংশনটি দেখুন: সেখানে model_cls.model_json_schema() কল করার সাথে সাথেই Pydantic নিজে থেকে আমাদের ক্লাসগুলোকে LLM-এর উপযোগী JSON Schema ডিকশনারিতে রূপান্তর করে দিচ্ছে। হাত দিয়ে একটি লাইনও JSON Schema লিখতে হচ্ছে না!

ধাপ ৩: এজেন্টের মূল রিজনিং লুপ তৈরি করা (agent/runner.py)

LangChain বা CrewAI-এর মতো বড় ফ্রেমওয়ার্ক ব্যবহার করার আগে অন্তত একবার নিজের হাতে ৬০ লাইনের একটি র-এজেন্ট লুপ (Raw Agent Loop) লেখা উচিত। এতে আপনার চোখের সামনে পরিষ্কার হয়ে যাবে কীভাবে মডেল stop_reason == 'tool_use' সিগন্যাল দেয় এবং কীভাবে একটি টুলের ফলাফল আবার messages লিস্টে ঢুকে পরের ধাপের চিন্তা শুরু করে।

চলুন agent/runner.py ফাইলটি তৈরি করি:

`python

agent/runner.py

import os import time from typing import Any, Dict, List, Literal, Optional from anthropic import Anthropic from pydantic import BaseModel from agent.tools import dispatch_agent_tool, get_anthropic_tool_schemas

SYSTEM_PROMPT = ( 'You are DevPulse, a technical triage and documentation AI agent.\n' '1. Always check internal documentation (search_knowledge_base) or live GitHub issues ' '(inspect_live_github_issue) before answering technical questions.\n' '2. Never invent error codes or issue IDs.\n' '3. Only call draft_issue_ticket when asked to log a confirmed bug.' )

class StepTrace(BaseModel): step_index: int thought_summary: str selected_tool: Optional[str] = None tool_input: Optional[Dict[str, Any]] = None tool_output_json: Optional[str] = None duration_ms: int

class AgentRunResult(BaseModel): success: bool final_answer: str steps: List[StepTrace] input_tokens: int output_tokens: int total_duration_ms: int halted_reason: Literal['completed', 'max_steps_reached', 'error']

def run_devpulse_python_agent( user_prompt: str, max_steps: int = 5, model_id: str = 'claude-3-5-sonnet-latest', ) -> AgentRunResult: start_time = time.perf_counter() api_key = os.environ.get('ANTHROPIC_API_KEY') if not api_key: raise RuntimeError('Missing ANTHROPIC_API_KEY in environment.')

client = Anthropic(api_key=api_key)
tools = get_anthropic_tool_schemas()
messages: List[Dict[str, Any]] = [
    {'role': 'user', 'content': user_prompt}
]

steps_trace: List[StepTrace] = []
total_in_tokens = 0
total_out_tokens = 0

for current_step in range(1, max_steps + 1):
    step_start = time.perf_counter()
    response = client.messages.create(
        model=model_id,
        max_tokens=1024,
        temperature=0.1,
        system=SYSTEM_PROMPT,
        tools=tools,
        messages=messages,
    )

    total_in_tokens += response.usage.input_tokens
    total_out_tokens += response.usage.output_tokens

    text_blocks = [
        b.text for b in response.content if getattr(b, 'type', None) == 'text'
    ]
    thought_text = ' '.join(text_blocks).strip() or f'Step {current_step} evaluation'

    if response.stop_reason != 'tool_use':
        steps_trace.append(
            StepTrace(
                step_index=current_step,
                thought_summary=thought_text[:200],
                duration_ms=int((time.perf_counter() - step_start) * 1000),
            )
        )
        return AgentRunResult(
            success=True,
            final_answer=thought_text,
            steps=steps_trace,
            input_tokens=total_in_tokens,
            output_tokens=total_out_tokens,
            total_duration_ms=int((time.perf_counter() - start_time) * 1000),
            halted_reason='completed',
        )

    messages.append({'role': 'assistant', 'content': response.content})
    tool_results_content: List[Dict[str, Any]] = []

    for block in response.content:
        if getattr(block, 'type', None) == 'tool_use':
            tool_name = block.name
            tool_args = block.input if isinstance(block.input, dict) else {}
            tool_id = block.id

            observation_json = dispatch_agent_tool(tool_name, tool_args)

            steps_trace.append(
                StepTrace(
                    step_index=current_step,
                    thought_summary=thought_text[:200],
                    selected_tool=tool_name,
                    tool_input=tool_args,
                    tool_output_json=observation_json,
                    duration_ms=int((time.perf_counter() - step_start) * 1000),
                )
            )

            tool_results_content.append({
                'type': 'tool_result',
                'tool_use_id': tool_id,
                'content': observation_json,
            })

    messages.append({'role': 'user', 'content': tool_results_content})

return AgentRunResult(
    success=False,
    final_answer='Reached maximum reasoning step limit before finalizing answer.',
    steps=steps_trace,
    input_tokens=total_in_tokens,
    output_tokens=total_out_tokens,
    total_duration_ms=int((time.perf_counter() - start_time) * 1000),
    halted_reason='max_steps_reached',
)

`

এই লুপটির তিনটি সূক্ষ্ম দিক খেয়াল করুন:

১. প্যারালাল টুল কল হ্যান্ডলিং (for block in response.content): মডেল যদি ১ নম্বর ধাপেই একসাথে দুটি টুল (search_knowledge_base এবং inspect_live_github_issue) কল করে, তবে আমাদের ভেতরের লুপটি দুটি টুলই রান করে তাদের নিজ নিজ tool_use_id মিলিয়ে একটি সিঙ্গেল লিস্টে (tool_results_content) জমা করে পরের ধাপে পাঠায়। ২. অসীম লুপ রোধ (range(1, max_steps + 1)): কোনো কারণে টুল বারবার ফেইল করলেও এজেন্ট ৫ ধাপের বেশি ঘুরবে না। ৩. টাইপড আউটপুট (AgentRunResult): পুরো রান শেষে কত ইনপুট/আউটপুট টোকেন খরচ হলো এবং কোন ধাপে কত মিলিসেকেন্ড লাগল তা একটি পরিচ্ছন্ন Pydantic অবজেক্ট হিসেবে রিটার্ন হয়—যা পরের পর্বে আমরা সরাসরি FastAPI থেকে JSON হিসেবে পাঠাতে পারব!

ধাপ ৪: টার্মিনালে এজেন্ট রান করে পরীক্ষা করা (main.py)

এবার প্রজেক্টের রুটে main.py ফাইলটি তৈরি করে আমাদের Python এজেন্টটিকে একটি যৌগিক প্রশ্ন দিয়ে চালিয়ে দেখি:

`python

main.py

from dotenv import load_dotenv from agent.runner import run_devpulse_python_agent

def main() -> None: load_dotenv() query = ( 'Check our internal rate-limit docs for 429 burst rules, ' 'and inspect live GitHub issue #402 in vercel/ai to see its current status.' )

print(f'Running DevPulse Python Agent...\nQuery: {query}\n')
result = run_devpulse_python_agent(user_prompt=query, max_steps=5)

for step in result.steps:
    tool_label = step.selected_tool or 'Final Synthesis'
    print(f'[Step {step.step_index}] {tool_label} ({step.duration_ms}ms)')
    if step.tool_input:
        print(f'  Args: {step.tool_input}')

print('\n--- Final Agent Answer ---')
print(result.final_answer)
print(
    f'\nTelemetry: {result.input_tokens} in / {result.output_tokens} out tokens '
    f'| Total Time: {result.total_duration_ms}ms | Status: {result.halted_reason}'
)

if name == 'main': main() `

জাহিদের বাস্তব অভিজ্ঞতা: Python এজেন্টে সাধারণ dict ব্যবহার করতে গিয়ে যেভাবে পুরো স্ক্রিপ্ট ক্র্যাশ করেছিল!

আমি যখন আমার মেশিন লার্নিং ও ডিপ লার্নিং প্রজেক্টগুলো (যেমন Gold Price Predictor এবং কম্পিউটার ভিশন রিসার্চ) করার পাশাপাশি প্রথমবার Python দিয়ে স্বতন্ত্র AI Agent স্ক্রিপ্ট লেখা শুরু করি, তখন শুরুতে অলসতা করে কোনো Pydantic মডেল ছাড়াই সাধারণ Python ডিকশনারি (dict) দিয়ে টুল আর্গুমেন্ট পার্স করছিলাম। আমি হাত দিয়ে একটি বড় JSON Schema ডিকশনারি লিখেছিলাম এবং টুল ফাংশনের ভেতরে সরাসরি repo = tool_args['repository'] ও limit = int(tool_args['top_k']) লিখে রেখেছিলাম।

কয়েকবার টেস্ট করার পর হঠাৎ একটি জটিল প্রশ্নে মডেল ভুল করে top_k-এর মান সংখ্যা (3) না পাঠিয়ে ইংরেজি শব্দে 'three' পাঠিয়ে বসল এবং আরেকটি অপশনাল কি বাদ দিয়ে দিল। যেহেতু মাঝখানে কোনো ভ্যালিডেশন লেয়ার ছিল না, তাই int('three') লাইনটিতে এসে সাথে সাথে ValueError ও KeyError খেয়ে আমার পুরো Python প্রসেসটি মাঝপথে ক্র্যাশ করল! তার চেয়েও বিরক্তিকর ব্যাপার ছিল—যখনই আমি ফাংশনে নতুন কোনো প্যারামিটার যোগ করতাম, আলাদা করে লিখে রাখা JSON Schema ডিকশনারি আপডেট করতে ভুলে যেতাম।

এরপর যখন আমি Pydantic v2 (BaseModel) ব্যবহার শুরু করলাম, তখন দুটি সমস্যাই এক নিমেষে গায়েব হয়ে গেল। model_cls.model_json_schema() ব্যবহারের কারণে আমার Python ক্লাস এবং মডেলের প্রম্পট স্কিমা সবসময় ১০০% সিঙ্ক থাকে। আর dispatch_agent_tool()-এর ভেতরে ValidationError ক্যাচ করে এরর মেসেজটি আবার মডেলের কাছে ফিরিয়ে দেওয়ায় মডেল পরের ধাপে নিজেই 'three'-কে 3 বানিয়ে কাজ শেষ করে ফেলে।

সাধারণ ভুল ও সতর্কতা: Python দিয়ে AI Agent বানানোর সময়

TypeScript থেকে Python-এ এজেন্ট তৈরি করার সময় নিচের ৪টি সাধারণ ভুল অবশ্যই এড়িয়ে চলবেন:

১. টাইমআউট ছাড়া requests.get() ব্যবহার করা (Missing Network Timeout): Python-এর পুরনো requests লাইব্রেরিতে ডিফল্টভাবে কোনো টাইমআউট থাকে না—ফলে বাইরের কোনো API সাড়া না দিলে আপনার এজেন্ট অনন্তকাল ঝুলে থাকবে। সবসময় httpx.Client(timeout=5.0) ব্যবহার করবেন। ২. হাত দিয়ে JSON Schema ডিকশনারি লেখা (Manual Schema Drift): কখনোই ম্যানুয়ালি ৩০ লাইনের JSON Schema ডিকশনারি লিখবেন না। এতে টাইপ ভুল হওয়ার সম্ভাবনা থাকে। সবসময় Pydantic BaseModel ও Field() লিখে .model_json_schema() কল করবেন। ৩. একাধিক প্যারালাল টুল কলের মধ্যে শুধু প্রথমটির উত্তর পাঠানো (Incomplete Tool Results): মডেল যদি এক টার্নে দুটি tool_use ব্লক পাঠায়, তবে পরের মেসেজে অবশ্যই দুটি tool_use_id-এর জন্যই tool_result পাঠাতে হবে। শুধু একটি টুল রান করে লুপ ব্রেক করলে API সাথে সাথে 400 Bad Request এরর দেবে। ৪. টুলের ভেতরের Exception আনহ্যান্ডেলড রেখে দেওয়া (Uncaught Tool Exceptions): কোনো ইস্যু না পাওয়া গেলে (404) বা ভ্যালিডেশন ফেইল করলে ফাংশন যেন পুরো লুপ ক্র্যাশ না করায়। সবসময় try/except দিয়ে ToolResultEnvelope(ok=False, agent_instruction=...) রিটার্ন করবেন।

প্রায়শই জিজ্ঞাসিত প্রশ্নাবলী (FAQ)

শুরুতেই LangChain বা CrewAI ব্যবহার না করে আমরা কেন অফিশিয়াল SDK ও Pydantic দিয়ে এই লুপটি বানালাম?

মাত্র ৬০ লাইনের এই নেটিভ for লুপটি নিজের হাতে লিখলে আপনার কাছে পরিষ্কার হয়ে যাবে কীভাবে stop_reason == 'tool_use' কাজ করে, কীভাবে প্যারালাল tool_use_id মেলাতে হয় এবং কীভাবে মেসেজ হিস্ট্রিতে টুলের আউটপুট যুক্ত হয়। এই ভিত্তিটি পরিষ্কার থাকলে একদিকে যেমন ছোট ও মাঝারি প্রোডাকশন সার্ভিসে কোনো ভারী ফ্রেমওয়ার্কের প্রয়োজনই হয় না, অন্যদিকে ভবিষ্যতে LangGraph বা CrewAI ডিবাগ করাও পানির মতো সহজ হয়ে যায়।

আমাদের Next.js প্রজেক্টের Supabase ডাটাবেস কি এই Python এজেন্ট থেকেও ব্যবহার করা যাবে?

হ্যাঁ, শতভাগ করা যাবে! ২৪তম পর্বে আমরা যে agent_sessions এবং agent_messages টেবিল তৈরি করেছিলাম, সেগুলো স্ট্যান্ডার্ড PostgreSQL টেবিল। ফলে Python-এর supabase বা psycopg লাইব্রেরি দিয়ে এই এজেন্ট থেকেও একই ডাটাবেস ব্যবহার করা যাবে—এমনকি ২৯তম পর্বে আমরা সেখানে pgvector যুক্ত করব।

টার্মিনালে চলা এই Python এজেন্টটিকে কীভাবে একটি আসল Backend API বানানো যায় যাতে ওয়েব বা মোবাইল অ্যাপ কল করতে পারে?

খুবই চমৎকার প্রশ্ন! বর্তমানে আমাদের run_devpulse_python_agent() ফাংশনটি একটি স্বতন্ত্র মডিউল হিসেবে তৈরি আছে। ঠিক পরের পর্বেই (পর্ব ২৮) আমরা FastAPI ব্যবহার করে এই Python এজেন্টটিকে একটি হাই-পারফরম্যান্স অ্যাসিঙ্ক্রোনাস REST ও স্ট্রিমিং ব্যাকএন্ড API-তে রূপান্তর করব।