هوش مصنوعی با Django؛ آموزش ساخت API، چت‌بات و سرویس AI با جنگو

Django یکی از محبوب‌ترین فریم‌ورک‌های توسعه وب با پایتون (Python) است. اگر وب‌سایت، فروشگاه اینترنتی، پنل سازمانی، سامانه آموزشی یا نرم‌افزار تحت وب شما با Django ساخته شده باشد، می‌توانید قابلیت‌های هوش مصنوعی را بدون بازنویسی معماری اصلی به آن اضافه کنید.

برای این کار لازم نیست مدل هوش مصنوعی را روی سرور خود آموزش دهید یا زیرساخت GPU راه‌اندازی کنید. Backend جنگو می‌تواند از طریق یک API به مدل انتخابی متصل شود، پرامپت (Prompt) را ارسال کند و پاسخ را به‌صورت متن، JSON یا Streaming دریافت کند.

در این مقاله یک پروژه واقعی Django می‌سازیم و آن را به API درواره متصل می‌کنیم.

پروژه نهایی قابلیت‌های زیر را خواهد داشت:

• ارسال پیام به مدل هوش مصنوعی
• دریافت پاسخ متنی
• دریافت پاسخ Streaming با Server-Sent Events
• خلاصه‌سازی متن
• تبدیل بازخورد مشتری به JSON
• ذخیره تاریخچه مکالمه در پایگاه داده
• مدیریت Timeout و خطاهای API
• ثبت میزان مصرف توکن (Token)
• تست Endpointها
• اجرای پروژه تحت ASGI
• استقرار با Docker

چرا Django برای ساخت برنامه‌های هوش مصنوعی مناسب است؟

Django معمولاً برای آموزش مدل‌های بزرگ استفاده نمی‌شود؛ اما برای ساخت لایه Backend محصولات مبتنی بر هوش مصنوعی انتخاب بسیار مناسبی است.

مهم‌ترین مزایای Django عبارت‌اند از:

• سیستم ORM قدرتمند
• پنل مدیریت داخلی
• سیستم کاربران و احراز هویت
• مدیریت Migration پایگاه داده
• Routing و Viewهای ساختاریافته
• پشتیبانی از Middleware
• قابلیت استفاده با PostgreSQL، MySQL و SQLite
• پشتیبانی از Async View و ASGI
• اکوسیستم بزرگ Packageهای پایتون
• مناسب برای پروژه‌های کوچک تا سامانه‌های سازمانی

Django می‌تواند مسئول بخش‌هایی باشد که مدل هوش مصنوعی به‌تنهایی انجام نمی‌دهد:

• مدیریت کاربران
• ذخیره مکالمات
• کنترل دسترسی
• محدودیت مصرف
• مدیریت اشتراک
• ثبت هزینه
• اتصال به پایگاه داده
• اجرای قوانین کسب‌وکار
• ارتباط با Frontend
• مدیریت فایل و سند

با Django و هوش مصنوعی چه برنامه‌هایی می‌توان ساخت؟

ترکیب Django و API هوش مصنوعی در پروژه‌های مختلفی کاربرد دارد.

معماری صحیح اتصال Django به هوش مصنوعی

در یک محصول واقعی، Frontend نباید مستقیماً به API مدل متصل شود. API Key باید فقط در Backend نگه‌داری شود.

جریان درخواست پیشنهادی به این صورت است:

1. کاربر پیام را در وب‌سایت یا اپلیکیشن وارد می‌کند.
2. Frontend پیام را به Backend جنگو می‌فرستد.
3. Django کاربر و ورودی را اعتبارسنجی می‌کند.
4. Django درخواست را با API Key محرمانه به درواره ارسال می‌کند.
5. درواره درخواست را به مدل انتخابی منتقل می‌کند.
6. پاسخ مدل به Django برمی‌گردد.
7. Django پاسخ و میزان مصرف را ثبت می‌کند.
8. پاسخ کنترل‌شده به Frontend ارسال می‌شود.

با این معماری می‌توانید مدل را تغییر دهید، محدودیت مصرف تعریف کنید، پاسخ‌ها را Cache کنید و تاریخچه مکالمات را در پایگاه داده نگه دارید.

پیش‌نیازهای آموزش

برای انجام این پروژه به موارد زیر نیاز دارید:

• یک نسخه جدید و پشتیبانی‌شده از Python
• یک نسخه پشتیبانی‌شده از Django
• آشنایی مقدماتی با Python و Django
• حساب کاربری در درواره
• API Key درواره
• شناسه یک مدل متنی

برای بررسی نسخه Python:

python --version

در بعضی سیستم‌ها باید از دستور زیر استفاده کنید:

python3 --version

اطلاعات اتصال در این آموزش:

Base URL:
https://api.darvareh.ir/v1

API Key:
YOUR_DARVAREH_API_KEY

Model ID:
YOUR_MODEL_ID

شناسه مدل و قیمت به‌روز آن را از صفحه مدل‌های درواره دریافت کنید.

ساخت محیط مجازی Python

پوشه پروژه را ایجاد کنید:

mkdir darvareh-django-ai
cd darvareh-django-ai

محیط مجازی بسازید:

python -m venv .venv

فعال‌کردن محیط مجازی در Linux و macOS:

source .venv/bin/activate

فعال‌کردن در Windows PowerShell:

.venv\Scripts\Activate.ps1

نصب Django و HTTPX

Packageهای موردنیاز را نصب کنید:

pip install django httpx uvicorn

برای محیط Production می‌توانید Uvicorn را با وابستگی‌های استاندارد نصب کنید:

pip install "uvicorn[standard]"

فایل وابستگی‌ها را بسازید:

pip freeze > requirements.txt

در این آموزش از HTTPX استفاده می‌کنیم؛ زیرا علاوه بر API هم‌زمان، از درخواست‌های Async و پاسخ Streaming پشتیبانی می‌کند. HTTPX یک HTTP Client برای Python با پشتیبانی از HTTP/1.1، HTTP/2، Timeout و Connection Pool است. جزئیات آن در مستندات رسمی HTTPX ارائه شده است.

ساخت پروژه Django

پروژه را در پوشه فعلی ایجاد کنید:

django-admin startproject config .

اپلیکیشن مربوط به قابلیت هوش مصنوعی را بسازید:

python manage.py startapp ai_chat

ساختار پروژه:

darvareh-django-ai/
├── ai_chat/
│   ├── migrations/
│   ├── services/
│   │   ├── __init__.py
│   │   └── darvareh.py
│   ├── __init__.py
│   ├── admin.py
│   ├── apps.py
│   ├── models.py
│   ├── tests.py
│   ├── urls.py
│   └── views.py
├── config/
│   ├── __init__.py
│   ├── asgi.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── manage.py
└── requirements.txt

پوشه services را ایجاد کنید:

mkdir ai_chat/services

سپس فایل خالی زیر را بسازید:

ai_chat/services/__init__.py

ثبت اپلیکیشن در تنظیمات

فایل config/settings.py را باز کنید و ai_chat را به INSTALLED_APPS اضافه کنید:

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "ai_chat",
]

تعریف تنظیمات درواره

در ابتدای config/settings.py ماژول os را Import کنید:

import os

سپس تنظیمات زیر را به انتهای فایل اضافه کنید:

DARVAREH_BASE_URL = os.getenv(
    "DARVAREH_BASE_URL",
    "https://api.darvareh.ir/v1",
)

DARVAREH_API_KEY = os.getenv(
    "DARVAREH_API_KEY",
    "",
)

DARVAREH_MODEL = os.getenv(
    "DARVAREH_MODEL",
    "",
)

DARVAREH_TIMEOUT_SECONDS = int(
    os.getenv("DARVAREH_TIMEOUT_SECONDS", "90")
)

API Key را مستقیماً در settings.py قرار ندهید.

روش نامناسب:

DARVAREH_API_KEY = "sk-..."

روش مناسب:

DARVAREH_API_KEY = os.getenv(
    "DARVAREH_API_KEY",
    "",
)

تنظیم Environment Variable

در Linux یا macOS:

export DARVAREH_API_KEY="YOUR_DARVAREH_API_KEY"
export DARVAREH_MODEL="YOUR_MODEL_ID"
export DARVAREH_BASE_URL="https://api.darvareh.ir/v1"

در Windows PowerShell:

$env:DARVAREH_API_KEY="YOUR_DARVAREH_API_KEY"
$env:DARVAREH_MODEL="YOUR_MODEL_ID"
$env:DARVAREH_BASE_URL="https://api.darvareh.ir/v1"

اگر در محیط توسعه از فایل .env استفاده می‌کنید، آن را به .gitignore اضافه کنید:

.env
.venv/
__pycache__/
*.pyc
db.sqlite3

ساخت Client اتصال به API درواره

فایل ai_chat/services/darvareh.py را ایجاد کنید:

import json
from collections.abc import AsyncIterator
from typing import Any

import httpx
from django.conf import settings


class DarvarehAPIError(Exception):
    def __init__(
        self,
        message: str,
        status_code: int | None = None,
    ):
        super().__init__(message)
        self.status_code = status_code


class DarvarehClient:
    def __init__(self) -> None:
        if not settings.DARVAREH_API_KEY:
            raise ValueError(
                "DARVAREH_API_KEY has not been configured."
            )

        if not settings.DARVAREH_MODEL:
            raise ValueError(
                "DARVAREH_MODEL has not been configured."
            )

        timeout = httpx.Timeout(
            timeout=settings.DARVAREH_TIMEOUT_SECONDS,
            connect=10.0,
        )

        limits = httpx.Limits(
            max_connections=100,
            max_keepalive_connections=20,
            keepalive_expiry=30.0,
        )

        self.model = settings.DARVAREH_MODEL

        self.http_client = httpx.AsyncClient(
            base_url=(
                settings.DARVAREH_BASE_URL.rstrip("/") + "/"
            ),
            headers={
                "Authorization": (
                    f"Bearer {settings.DARVAREH_API_KEY}"
                ),
                "Accept": "application/json",
                "Content-Type": "application/json",
            },
            timeout=timeout,
            limits=limits,
        )

    async def complete(
        self,
        message: str,
        system_prompt: str | None = None,
        *,
        temperature: float = 0.3,
        max_tokens: int = 1000,
    ) -> dict[str, Any]:
        messages = self._build_messages(
            message,
            system_prompt,
        )

        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
        }

        try:
            response = await self.http_client.post(
                "chat/completions",
                json=payload,
            )
            response.raise_for_status()
        except httpx.TimeoutException as exc:
            raise DarvarehAPIError(
                "The AI request timed out."
            ) from exc
        except httpx.HTTPStatusError as exc:
            raise self._create_status_error(
                exc.response
            ) from exc
        except httpx.RequestError as exc:
            raise DarvarehAPIError(
                "Could not connect to the AI service."
            ) from exc

        try:
            data = response.json()
            choice = data["choices"][0]
            answer = choice["message"]["content"]
        except (
            ValueError,
            KeyError,
            IndexError,
            TypeError,
        ) as exc:
            raise DarvarehAPIError(
                "The AI response had an unexpected structure."
            ) from exc

        if not isinstance(answer, str) or not answer.strip():
            raise DarvarehAPIError(
                "The AI response did not contain any text."
            )

        return {
            "answer": answer,
            "model": data.get("model", self.model),
            "usage": data.get("usage"),
        }

    async def complete_json(
        self,
        prompt: str,
        *,
        max_tokens: int = 600,
    ) -> dict[str, Any]:
        payload = {
            "model": self.model,
            "messages": [
                {
                    "role": "system",
                    "content": (
                        "فقط یک JSON معتبر تولید کن. "
                        "هیچ متن، توضیح یا Markdown "
                        "خارج از JSON ننویس."
                    ),
                },
                {
                    "role": "user",
                    "content": prompt,
                },
            ],
            "temperature": 0,
            "max_tokens": max_tokens,
            "response_format": {
                "type": "json_object",
            },
        }

        try:
            response = await self.http_client.post(
                "chat/completions",
                json=payload,
            )
            response.raise_for_status()
        except httpx.TimeoutException as exc:
            raise DarvarehAPIError(
                "The JSON request timed out."
            ) from exc
        except httpx.HTTPStatusError as exc:
            raise self._create_status_error(
                exc.response
            ) from exc
        except httpx.RequestError as exc:
            raise DarvarehAPIError(
                "Could not connect to the AI service."
            ) from exc

        try:
            data = response.json()
            content = data["choices"][0]["message"]["content"]
            parsed = json.loads(content)
        except (
            ValueError,
            KeyError,
            IndexError,
            TypeError,
            json.JSONDecodeError,
        ) as exc:
            raise DarvarehAPIError(
                "The model did not return valid JSON."
            ) from exc

        if not isinstance(parsed, dict):
            raise DarvarehAPIError(
                "The model JSON must be an object."
            )

        return parsed

    async def stream(
        self,
        message: str,
        system_prompt: str | None = None,
        *,
        temperature: float = 0.3,
        max_tokens: int = 1000,
    ) -> AsyncIterator[str]:
        payload = {
            "model": self.model,
            "messages": self._build_messages(
                message,
                system_prompt,
            ),
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": True,
        }

        try:
            async with self.http_client.stream(
                "POST",
                "chat/completions",
                json=payload,
                headers={
                    "Accept": "text/event-stream",
                },
            ) as response:
                if not response.is_success:
                    body = await response.aread()

                    raise DarvarehAPIError(
                        (
                            "AI API returned status "
                            f"{response.status_code}: "
                            f"{body[:2000].decode(errors='replace')}"
                        ),
                        status_code=response.status_code,
                    )

                async for line in response.aiter_lines():
                    line = line.strip()

                    if not line.startswith("data:"):
                        continue

                    data = line.removeprefix(
                        "data:"
                    ).strip()

                    if data == "[DONE]":
                        return

                    try:
                        event = json.loads(data)
                        choices = event.get("choices", [])

                        if not choices:
                            continue

                        token = (
                            choices[0]
                            .get("delta", {})
                            .get("content")
                        )
                    except (
                        json.JSONDecodeError,
                        AttributeError,
                        IndexError,
                    ):
                        continue

                    if isinstance(token, str) and token:
                        yield token

        except httpx.TimeoutException as exc:
            raise DarvarehAPIError(
                "The streaming request timed out."
            ) from exc
        except httpx.RequestError as exc:
            raise DarvarehAPIError(
                "The streaming connection failed."
            ) from exc

    async def close(self) -> None:
        await self.http_client.aclose()

    def _build_messages(
        self,
        message: str,
        system_prompt: str | None,
    ) -> list[dict[str, str]]:
        messages: list[dict[str, str]] = []

        if system_prompt and system_prompt.strip():
            messages.append({
                "role": "system",
                "content": system_prompt,
            })

        messages.append({
            "role": "user",
            "content": message,
        })

        return messages

    def _create_status_error(
        self,
        response: httpx.Response,
    ) -> DarvarehAPIError:
        body = response.text[:2000]

        return DarvarehAPIError(
            (
                "AI API returned status "
                f"{response.status_code}: {body}"
            ),
            status_code=response.status_code,
        )


darvareh_client = DarvarehClient()

در کد بالا یک نمونه مشترک از httpx.AsyncClient ایجاد شده است. این کار امکان استفاده از Connection Pool را فراهم می‌کند.

مستندات HTTPX توصیه می‌کند در مسیرهای پرتکرار برای هر درخواست یک Client جدید ساخته نشود؛ زیرا استفاده مجدد از Client باعث استفاده بهتر از Connection Pool می‌شود. توضیحات بیشتر در راهنمای Async HTTPX موجود است.

چرا از Async Client استفاده می‌کنیم؟

ارتباط با مدل هوش مصنوعی یک عملیات I/O است. بیشتر زمان درخواست صرف انتظار برای پاسخ شبکه می‌شود.

Django از Async View پشتیبانی می‌کند و در حالت ASGI می‌تواند درخواست‌های طولانی و Streaming را بدون اختصاص یک Thread مجزا به هر اتصال مدیریت کند.

طبق مستندات رسمی Async در Django، برای بهره‌بردن از زنجیره کاملاً Async و مدیریت مناسب اتصال‌های طولانی باید پروژه تحت ASGI اجرا شود.

ساخت Viewهای API

فایل ai_chat/views.py را به شکل زیر بنویسید:

import json
import logging
from typing import Any

from django.http import (
    HttpRequest,
    JsonResponse,
    StreamingHttpResponse,
)
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import (
    require_GET,
    require_POST,
)

from .services.darvareh import (
    DarvarehAPIError,
    darvareh_client,
)


logger = logging.getLogger(__name__)


def parse_json_body(
    request: HttpRequest,
    *,
    max_bytes: int = 32_000,
) -> dict[str, Any]:
    content_length = request.headers.get(
        "Content-Length"
    )

    if content_length:
        try:
            if int(content_length) > max_bytes:
                raise ValueError(
                    "Request body is too large."
                )
        except ValueError as exc:
            if str(exc) == "Request body is too large.":
                raise

    if len(request.body) > max_bytes:
        raise ValueError(
            "Request body is too large."
        )

    try:
        data = json.loads(request.body)
    except json.JSONDecodeError as exc:
        raise ValueError(
            "Request body must be valid JSON."
        ) from exc

    if not isinstance(data, dict):
        raise ValueError(
            "JSON body must be an object."
        )

    return data


def validate_text(
    value: Any,
    *,
    name: str,
    minimum: int = 1,
    maximum: int = 12_000,
    required: bool = True,
) -> str | None:
    if value is None and not required:
        return None

    if not isinstance(value, str):
        raise ValueError(
            f"{name} must be a string."
        )

    cleaned = value.strip()

    if len(cleaned) < minimum:
        raise ValueError(
            f"{name} is too short."
        )

    if len(cleaned) > maximum:
        raise ValueError(
            f"{name} is too long."
        )

    return cleaned


@require_GET
async def health(request: HttpRequest) -> JsonResponse:
    return JsonResponse({
        "status": "ok",
    })


@csrf_exempt
@require_POST
async def chat(request: HttpRequest) -> JsonResponse:
    try:
        data = parse_json_body(request)

        message = validate_text(
            data.get("message"),
            name="message",
            maximum=12_000,
        )

        system_prompt = validate_text(
            data.get("system_prompt"),
            name="system_prompt",
            maximum=2_000,
            required=False,
        )
    except ValueError as exc:
        return JsonResponse(
            {
                "error": str(exc),
            },
            status=400,
        )

    try:
        result = await darvareh_client.complete(
            message=message,
            system_prompt=system_prompt,
        )
    except DarvarehAPIError as exc:
        logger.exception(
            "AI completion failed"
        )

        return JsonResponse(
            {
                "error": (
                    "دریافت پاسخ از سرویس "
                    "هوش مصنوعی ناموفق بود."
                ),
            },
            status=502,
        )

    return JsonResponse(result)


@csrf_exempt
@require_POST
async def summarize(
    request: HttpRequest,
) -> JsonResponse:
    try:
        data = parse_json_body(
            request,
            max_bytes=64_000,
        )

        text = validate_text(
            data.get("text"),
            name="text",
            minimum=20,
            maximum=30_000,
        )

        max_paragraphs = data.get(
            "max_paragraphs",
            3,
        )

        if (
            not isinstance(max_paragraphs, int)
            or isinstance(max_paragraphs, bool)
            or max_paragraphs < 1
            or max_paragraphs > 10
        ):
            raise ValueError(
                "max_paragraphs must be between 1 and 10."
            )
    except ValueError as exc:
        return JsonResponse(
            {
                "error": str(exc),
            },
            status=400,
        )

    prompt = f"""
متن زیر را حداکثر در {max_paragraphs} پاراگراف خلاصه کن.

الزامات:
- اطلاعات اصلی حفظ شوند.
- مطلب جدیدی اضافه نشود.
- پاسخ فارسی روان باشد.
- از تکرار پرهیز شود.

متن:
{text}
""".strip()

    try:
        result = await darvareh_client.complete(
            message=prompt,
            system_prompt=(
                "تو یک ویراستار دقیق فارسی هستی."
            ),
            temperature=0.2,
            max_tokens=1_000,
        )
    except DarvarehAPIError:
        logger.exception(
            "AI summarization failed"
        )

        return JsonResponse(
            {
                "error": "خلاصه‌سازی متن ناموفق بود.",
            },
            status=502,
        )

    return JsonResponse(result)


@csrf_exempt
@require_POST
async def analyze_feedback(
    request: HttpRequest,
) -> JsonResponse:
    try:
        data = parse_json_body(request)

        text = validate_text(
            data.get("text"),
            name="text",
            minimum=3,
            maximum=5_000,
        )
    except ValueError as exc:
        return JsonResponse(
            {
                "error": str(exc),
            },
            status=400,
        )

    prompt = f"""
بازخورد زیر را تحلیل کن:

{text}

دقیقاً این ساختار JSON را برگردان:
{{
  "category": "product | delivery | payment | support | other",
  "sentiment": "positive | neutral | negative",
  "priority": 1,
  "summary": "خلاصه کوتاه فارسی"
}}

priority باید عددی بین 1 تا 5 باشد.
""".strip()

    try:
        result = await darvareh_client.complete_json(
            prompt
        )

        validate_feedback_result(result)
    except (
        DarvarehAPIError,
        ValueError,
    ):
        logger.exception(
            "Feedback analysis failed"
        )

        return JsonResponse(
            {
                "error": (
                    "خروجی مدل با ساختار "
                    "مورد انتظار سازگار نبود."
                ),
            },
            status=502,
        )

    return JsonResponse(result)


def validate_feedback_result(
    result: dict[str, Any],
) -> None:
    valid_categories = {
        "product",
        "delivery",
        "payment",
        "support",
        "other",
    }

    valid_sentiments = {
        "positive",
        "neutral",
        "negative",
    }

    if result.get("category") not in valid_categories:
        raise ValueError(
            "Invalid feedback category."
        )

    if result.get("sentiment") not in valid_sentiments:
        raise ValueError(
            "Invalid feedback sentiment."
        )

    priority = result.get("priority")

    if (
        not isinstance(priority, int)
        or isinstance(priority, bool)
        or priority < 1
        or priority > 5
    ):
        raise ValueError(
            "Invalid feedback priority."
        )

    summary = result.get("summary")

    if (
        not isinstance(summary, str)
        or not summary.strip()
    ):
        raise ValueError(
            "Feedback summary is empty."
        )


@csrf_exempt
@require_POST
async def stream_chat(
    request: HttpRequest,
) -> StreamingHttpResponse | JsonResponse:
    try:
        data = parse_json_body(request)

        message = validate_text(
            data.get("message"),
            name="message",
            maximum=12_000,
        )

        system_prompt = validate_text(
            data.get("system_prompt"),
            name="system_prompt",
            maximum=2_000,
            required=False,
        )
    except ValueError as exc:
        return JsonResponse(
            {
                "error": str(exc),
            },
            status=400,
        )

    async def event_stream():
        try:
            async for token in darvareh_client.stream(
                message=message,
                system_prompt=system_prompt,
            ):
                payload = json.dumps(
                    {
                        "token": token,
                    },
                    ensure_ascii=False,
                )

                yield f"data: {payload}\n\n"

            yield "data: [DONE]\n\n"

        except DarvarehAPIError:
            logger.exception(
                "AI streaming failed"
            )

            payload = json.dumps(
                {
                    "error": "stream_failed",
                },
                ensure_ascii=False,
            )

            yield (
                "event: error\n"
                f"data: {payload}\n\n"
            )

    response = StreamingHttpResponse(
        event_stream(),
        content_type="text/event-stream",
    )

    response["Cache-Control"] = "no-cache"
    response["X-Accel-Buffering"] = "no"

    return response

در نمونه آموزشی از csrf_exempt استفاده شده تا Endpointها با cURL و Clientهای غیرمرورگری قابل‌آزمایش باشند. در محصول واقعی، Endpointها باید با روش احراز هویت متناسب با معماری شما محافظت شوند و تصمیم درباره CSRF بر اساس نوع Session، Cookie و Client گرفته شود.

تعریف URLهای اپلیکیشن

فایل ai_chat/urls.py را ایجاد کنید:

from django.urls import path

from . import views


app_name = "ai_chat"


urlpatterns = [
    path(
        "health/",
        views.health,
        name="health",
    ),
    path(
        "chat/",
        views.chat,
        name="chat",
    ),
    path(
        "chat/stream/",
        views.stream_chat,
        name="stream-chat",
    ),
    path(
        "summarize/",
        views.summarize,
        name="summarize",
    ),
    path(
        "analyze-feedback/",
        views.analyze_feedback,
        name="analyze-feedback",
    ),
]

فایل config/urls.py:

from django.contrib import admin
from django.urls import include, path


urlpatterns = [
    path("admin/", admin.site.urls),
    path(
        "api/ai/",
        include("ai_chat.urls"),
    ),
]

Endpointهای نهایی:

GET  /api/ai/health/
POST /api/ai/chat/
POST /api/ai/chat/stream/
POST /api/ai/summarize/
POST /api/ai/analyze-feedback/

اجرای Migration

python manage.py migrate

برای بررسی تنظیمات پروژه:

python manage.py check

اجرای پروژه با سرور توسعه Django

برای تست پاسخ معمولی:

python manage.py runserver

آدرس پیش‌فرض:

http://127.0.0.1:8000

برای Streaming و محیط‌های Async بهتر است پروژه را تحت ASGI اجرا کنید.

اجرای Django با Uvicorn و ASGI

Django به‌صورت پیش‌فرض فایل config/asgi.py را ایجاد می‌کند.

پروژه را با Uvicorn اجرا کنید:

uvicorn config.asgi:application \
  --host 0.0.0.0 \
  --port 8000 \
  --reload

گزینه --reload فقط برای محیط توسعه مناسب است.

طبق مستندات StreamingHttpResponse جنگو، Streaming تحت ASGI می‌تواند بدون مسدودکردن یک Worker برای تمام مدت پاسخ، اتصال‌های طولانی مانند SSE را مدیریت کند. در WSGI هر پاسخ Streaming ممکن است Worker را تا پایان اتصال درگیر نگه دارد.

آزمایش Health Check

curl http://127.0.0.1:8000/api/ai/health/

پاسخ:

{
  "status": "ok"
}

آزمایش Endpoint چت

curl -X POST \
  "http://127.0.0.1:8000/api/ai/chat/" \
  -H "Content-Type: application/json" \
  -d '{
    "message": "Django ORM چیست و چه کاربردی دارد؟",
    "system_prompt": "پاسخ را ساده، دقیق و فارسی بنویس."
  }'

نمونه پاسخ:

{
  "answer": "Django ORM لایه‌ای برای ارتباط با پایگاه داده از طریق کلاس‌ها و اشیای پایتون است.",
  "model": "YOUR_MODEL_ID",
  "usage": {
    "prompt_tokens": 39,
    "completion_tokens": 61,
    "total_tokens": 100
  }
}

محتوا و میزان توکن واقعی به مدل و پرامپت بستگی دارند.

آزمایش خلاصه‌سازی متن

curl -X POST \
  "http://127.0.0.1:8000/api/ai/summarize/" \
  -H "Content-Type: application/json" \
  -d '{
    "text": "Django یک فریم‌ورک توسعه وب با زبان پایتون است که ابزارهایی برای مدیریت پایگاه داده، کاربران، فرم‌ها، پنل مدیریت و مسیریابی ارائه می‌دهد. این فریم‌ورک برای ساخت سریع برنامه‌های وب طراحی شده است.",
    "max_paragraphs": 2
  }'

آزمایش تحلیل بازخورد

curl -X POST \
  "http://127.0.0.1:8000/api/ai/analyze-feedback/" \
  -H "Content-Type: application/json" \
  -d '{
    "text": "کیفیت محصول خوب بود اما سفارش با چهار روز تأخیر رسید."
  }'

نمونه پاسخ:

{
  "category": "delivery",
  "sentiment": "negative",
  "priority": 3,
  "summary": "مشتری از تأخیر در تحویل سفارش ناراضی است."
}

پارامتر response_format باید توسط مدل انتخابی پشتیبانی شود. اگر مدل از JSON Mode پشتیبانی نمی‌کند، می‌توانید این پارامتر را حذف کنید؛ اما اعتبارسنجی خروجی همچنان ضروری است.

آزمایش Streaming با cURL

گزینه -N از Bufferشدن خروجی در cURL جلوگیری می‌کند:

curl -N -X POST \
  "http://127.0.0.1:8000/api/ai/chat/stream/" \
  -H "Content-Type: application/json" \
  -d '{
    "message": "Middleware در Django را توضیح بده.",
    "system_prompt": "با یک مثال ساده و به زبان فارسی پاسخ بده."
  }'

خروجی به‌تدریج دریافت می‌شود:

data: {"token": "Middleware"}

data: {"token": " در"}

data: {"token": " Django"}

data: {"token": " لایه‌ای..."}

data: [DONE]

دریافت Streaming در Frontend

در Frontend می‌توان پاسخ POST را با fetch و ReadableStream دریافت کرد:

async function streamChat(message, onToken) {
  const response = await fetch(
    "https://api.example.com/api/ai/chat/stream/",
    {
      method: "POST",
      headers: {
        "Content-Type": "application/json"
      },
      body: JSON.stringify({
        message,
        system_prompt: "پاسخ را فارسی بنویس."
      })
    }
  );

  if (!response.ok || !response.body) {
    throw new Error(
      `Request failed: ${response.status}`
    );
  }

  const reader = response.body.getReader();
  const decoder = new TextDecoder();

  let buffer = "";

  while (true) {
    const { value, done } = await reader.read();

    if (done) {
      break;
    }

    buffer += decoder.decode(value, {
      stream: true
    });

    const events = buffer.split("\n\n");
    buffer = events.pop() ?? "";

    for (const event of events) {
      const dataLine = event
        .split("\n")
        .find(line => line.startsWith("data:"));

      if (!dataLine) {
        continue;
      }

      const data = dataLine.slice(5).trim();

      if (data === "[DONE]") {
        return;
      }

      const parsed = JSON.parse(data);

      if (parsed.token) {
        onToken(parsed.token);
      }
    }
  }
}

let answer = "";

streamChat(
  "Django Signals چیست؟",
  token => {
    answer += token;

    document.querySelector(
      "#answer"
    ).textContent = answer;
  }
);

متغیر buffer ضروری است؛ زیرا هر Chunk شبکه لزوماً دقیقاً شامل یک رویداد کامل SSE نیست.

ذخیره تاریخچه مکالمه در Django

برای ساخت چت‌بات واقعی باید مکالمه‌ها و پیام‌ها را در پایگاه داده ذخیره کنید.

فایل ai_chat/models.py:

import uuid

from django.conf import settings
from django.db import models


class Conversation(models.Model):
    id = models.UUIDField(
        primary_key=True,
        default=uuid.uuid4,
        editable=False,
    )

    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        related_name="ai_conversations",
    )

    title = models.CharField(
        max_length=200,
        blank=True,
    )

    created_at = models.DateTimeField(
        auto_now_add=True,
    )

    updated_at = models.DateTimeField(
        auto_now=True,
    )

    class Meta:
        ordering = ["-updated_at"]

    def __str__(self) -> str:
        return self.title or str(self.id)


class Message(models.Model):
    class Role(models.TextChoices):
        SYSTEM = "system", "System"
        USER = "user", "User"
        ASSISTANT = "assistant", "Assistant"

    conversation = models.ForeignKey(
        Conversation,
        on_delete=models.CASCADE,
        related_name="messages",
    )

    role = models.CharField(
        max_length=20,
        choices=Role.choices,
    )

    content = models.TextField()

    model = models.CharField(
        max_length=200,
        blank=True,
    )

    prompt_tokens = models.PositiveIntegerField(
        null=True,
        blank=True,
    )

    completion_tokens = models.PositiveIntegerField(
        null=True,
        blank=True,
    )

    total_tokens = models.PositiveIntegerField(
        null=True,
        blank=True,
    )

    created_at = models.DateTimeField(
        auto_now_add=True,
    )

    class Meta:
        ordering = ["created_at"]

    def __str__(self) -> str:
        return (
            f"{self.role}: "
            f"{self.content[:50]}"
        )

Migration بسازید:

python manage.py makemigrations
python manage.py migrate

ثبت مدل‌ها در پنل مدیریت

فایل ai_chat/admin.py:

from django.contrib import admin

from .models import Conversation, Message


class MessageInline(admin.TabularInline):
    model = Message
    extra = 0
    readonly_fields = (
        "role",
        "content",
        "model",
        "prompt_tokens",
        "completion_tokens",
        "total_tokens",
        "created_at",
    )


@admin.register(Conversation)
class ConversationAdmin(admin.ModelAdmin):
    list_display = (
        "id",
        "user",
        "title",
        "created_at",
        "updated_at",
    )

    search_fields = (
        "title",
        "user__username",
    )

    inlines = [MessageInline]


@admin.register(Message)
class MessageAdmin(admin.ModelAdmin):
    list_display = (
        "id",
        "conversation",
        "role",
        "model",
        "total_tokens",
        "created_at",
    )

    list_filter = (
        "role",
        "model",
    )

    search_fields = (
        "content",
    )

ساخت کاربر مدیر:

python manage.py createsuperuser

سپس پنل مدیریت:

http://127.0.0.1:8000/admin/

ساخت تاریخچه مناسب برای مدل

برای ادامه مکالمه، پیام‌های قبلی باید در درخواست قرار بگیرند:

messages = [
    {
        "role": "system",
        "content": (
            "تو دستیار پشتیبانی فروشگاه هستی."
        ),
    },
    {
        "role": "user",
        "content": (
            "زمان ارسال سفارش چقدر است؟"
        ),
    },
    {
        "role": "assistant",
        "content": (
            "معمولاً دو تا چهار روز کاری."
        ),
    },
    {
        "role": "user",
        "content": "برای شهرستان چطور؟",
    },
]

اگر فقط آخرین پیام ارسال شود، مدل نمی‌داند عبارت «برای شهرستان چطور؟» به زمان ارسال اشاره می‌کند.

کنترل طول تاریخچه

ارسال تمام پیام‌های یک مکالمه طولانی باعث افزایش هزینه، زمان پاسخ و مصرف Context Window می‌شود.

یک سیاست عملی:

• System Prompt را نگه دارید.
• آخرین ۱۰ تا ۲۰ پیام را ارسال کنید.
• پیام‌های قدیمی‌تر را خلاصه کنید.
• داده‌های ثابت کاربر را جدا نگه دارید.
• پیام‌های خالی یا غیرضروری را حذف کنید.
• مجموع توکن‌ها را کنترل کنید.

ساختار پیشنهادی Context:

System Prompt
خلاصه پیام‌های قدیمی
اطلاعات ضروری کاربر
چند پیام اخیر
پیام جدید

استفاده از ORM در Async View

در نسخه‌های جدید Django بسیاری از عملیات ORM معادل Async دارند:

conversation = await Conversation.objects.acreate(
    user=request.user,
    title="گفت‌وگوی جدید",
)

ایجاد پیام:

await Message.objects.acreate(
    conversation=conversation,
    role=Message.Role.USER,
    content=message,
)

دریافت پیام‌ها:

messages = []

queryset = (
    Message.objects
    .filter(conversation=conversation)
    .order_by("-created_at")[:20]
)

async for item in queryset:
    messages.append({
        "role": item.role,
        "content": item.content,
    })

messages.reverse()

اگر بخشی از کد یا Package شما فقط Sync است، نباید آن را مستقیماً در Async View اجرا کنید. در آن حالت می‌توان از sync_to_async استفاده کرد:

from asgiref.sync import sync_to_async


result = await sync_to_async(
    some_sync_function
)()

ثبت مصرف توکن

پاسخ API ممکن است اطلاعاتی مشابه زیر داشته باشد:

{
  "usage": {
    "prompt_tokens": 120,
    "completion_tokens": 80,
    "total_tokens": 200
  }
}

این اطلاعات را برای هر پیام ذخیره کنید:

usage = result.get("usage") or {}

await Message.objects.acreate(
    conversation=conversation,
    role=Message.Role.ASSISTANT,
    content=result["answer"],
    model=result["model"],
    prompt_tokens=usage.get(
        "prompt_tokens"
    ),
    completion_tokens=usage.get(
        "completion_tokens"
    ),
    total_tokens=usage.get(
        "total_tokens"
    ),
)

ثبت مصرف برای این موارد مفید است:

• محاسبه هزینه هر قابلیت
• محدودکردن کاربران
• شناسایی پرامپت‌های پرهزینه
• مقایسه مدل‌ها
• طراحی بسته‌های اشتراک
• تشخیص افزایش ناگهانی مصرف

انتخاب Temperature مناسب

پارامتر temperature میزان تنوع پاسخ را کنترل می‌کند.

این اعداد نقطه شروع هستند. نتیجه واقعی به مدل انتخابی وابسته است و باید با داده‌های محصول آزمایش شود.

مدیریت خطاهای API

در Client چند گروه خطا مدیریت شده‌اند:

• Timeout
• خطای اتصال
• وضعیت HTTP ناموفق
• JSON نامعتبر
• پاسخ بدون متن
• ساختار پاسخ غیرمنتظره

نگاشت پیشنهادی خطاها:

جزئیات کامل خطای بالادستی را مستقیماً به کاربر نمایش ندهید. API Key یا Headerهای محرمانه نیز نباید در Log ثبت شوند.

مدیریت Timeout

در تنظیمات مقدار پیش‌فرض زیر را تعریف کردیم:

DARVAREH_TIMEOUT_SECONDS = 90

Timeout باید متناسب با کاربرد تنظیم شود:

برای یک کار طولانی، باز نگه‌داشتن درخواست HTTP برای چند دقیقه معمولاً انتخاب مناسبی نیست.

پردازش طولانی با Celery

برای عملیات طولانی بهتر است از Job Queue استفاده کنید.

جریان پیشنهادی:

1. کاربر درخواست را ثبت می‌کند.
2. Django یک Job می‌سازد.
3. Job وارد Queue می‌شود.
4. Celery Worker پردازش را انجام می‌دهد.
5. نتیجه در پایگاه داده ذخیره می‌شود.
6. کاربر با Job ID وضعیت را بررسی می‌کند.

نمونه پاسخ:

{
  "job_id": "job_72f9c",
  "status": "queued"
}

این معماری برای موارد زیر مناسب است:

• تحلیل تعداد زیادی سند
• خلاصه‌سازی فایل‌های طولانی
• تولید گروهی توضیحات محصول
• پردازش چندمرحله‌ای
• تولید گزارش بزرگ
• اجرای چند درخواست مدل

مدیریت Retry

Retry برای همه خطاها مناسب نیست.

در این موارد معمولاً نباید Retry انجام شود:

• API Key نامعتبر
• Model ID اشتباه
• ورودی نامعتبر
• Payload ناقص
• پاسخ 400
• پاسخ 401

Retry محدود می‌تواند برای این موارد مناسب باشد:

• بعضی خطاهای ۵xx
• قطع موقت اتصال
• Timeout اتصال
• وضعیت 429 با رعایت تأخیر

Backoff ساده:

import asyncio


async def wait_before_retry(
    attempt: int,
) -> None:
    delays = [0.5, 1, 2]

    delay = delays[
        min(attempt, len(delays) - 1)
    ]

    await asyncio.sleep(delay)

تعداد تلاش‌ها را محدود کنید و پیش از Retry بررسی کنید که عملیات تکراری اثر جانبی ایجاد نکند.

Rate Limiting

هر درخواست هوش مصنوعی هزینه ایجاد می‌کند. بنابراین Endpointها باید محدودیت مصرف داشته باشند.

Rate Limit را می‌توانید بر اساس موارد زیر اعمال کنید:

• User ID
• Organization ID
• نوع اشتراک
• API Key داخلی
• IP Address

سیاست نمونه:

کاربر عادی: 20 درخواست در دقیقه
کاربر حرفه‌ای: 100 درخواست در دقیقه
سازمان: بر اساس قرارداد

در پروژه تک‌سرور می‌توان از Cache داخلی استفاده کرد؛ اما در محیط چندسروری بهتر است وضعیت محدودیت در Redis نگه‌داری شود.

کاهش هزینه API هوش مصنوعی

برای کاهش هزینه در Django:

• طول پیام ورودی را محدود کنید.
• max_tokens را متناسب تنظیم کنید.
• تاریخچه قدیمی را خلاصه کنید.
• پاسخ‌های تکراری را Cache کنید.
• مدل مناسب همان وظیفه را انتخاب کنید.
• عملیات ساده را به مدل بزرگ نسپارید.
• ورودی نامعتبر را قبل از API رد کنید.
• مصرف هر کاربر را ثبت کنید.
• سقف روزانه و ماهانه تعریف کنید.
• نسخه پرامپت را ثبت کنید.

برای بررسی مدل‌ها و قیمت‌های به‌روز، صفحه مدل‌های درواره را ببینید.

کش‌کردن پاسخ‌ها

Django دارای Cache Framework داخلی است.

نمونه ساده:

import hashlib
import json

from django.core.cache import cache


def create_cache_key(
    *,
    model: str,
    system_prompt: str,
    message: str,
    temperature: float,
) -> str:
    payload = json.dumps(
        {
            "model": model,
            "system_prompt": system_prompt,
            "message": message,
            "temperature": temperature,
        },
        ensure_ascii=False,
        sort_keys=True,
    )

    digest = hashlib.sha256(
        payload.encode("utf-8")
    ).hexdigest()

    return f"ai-response:{digest}"

خواندن از Cache:

cache_key = create_cache_key(
    model=settings.DARVAREH_MODEL,
    system_prompt=system_prompt or "",
    message=message,
    temperature=0.3,
)

cached = await cache.aget(cache_key)

if cached is not None:
    return JsonResponse(cached)

ذخیره پاسخ:

await cache.aset(
    cache_key,
    result,
    timeout=3600,
)

Cache برای این موارد مناسب است:

• سؤال‌های متداول
• خلاصه سند ثابت
• طبقه‌بندی ورودی تکراری
• تولید توضیح ثابت

برای داده شخصی یا لحظه‌ای باید سیاست Cache با دقت طراحی شود.

نسخه‌بندی پرامپت‌ها

پرامپت‌های اصلی را به‌صورت متن پراکنده در Viewها نگه ندارید.

یک ساختار بهتر:

FEEDBACK_PROMPT_VERSION = "3"

FEEDBACK_SYSTEM_PROMPT = """
تو تحلیل‌گر بازخورد مشتری هستی.
فقط JSON معتبر تولید کن.
""".strip()

هنگام ذخیره نتیجه، این اطلاعات را ثبت کنید:

Prompt Name
Prompt Version
Model
Temperature
Max Tokens
Created At

در این صورت می‌توانید کیفیت نسخه‌های مختلف پرامپت را مقایسه کنید.

تست Endpoint چت

فایل ai_chat/tests.py:

import json
from unittest.mock import AsyncMock, patch

from django.test import TestCase
from django.urls import reverse


class ChatAPITests(TestCase):
    @patch(
        "ai_chat.views.darvareh_client.complete",
        new_callable=AsyncMock,
    )
    def test_chat_returns_ai_response(
        self,
        mock_complete,
    ):
        mock_complete.return_value = {
            "answer": "پاسخ آزمایشی",
            "model": "test-model",
            "usage": {
                "prompt_tokens": 10,
                "completion_tokens": 5,
                "total_tokens": 15,
            },
        }

        response = self.client.post(
            reverse("ai_chat:chat"),
            data=json.dumps({
                "message": "سلام",
            }),
            content_type="application/json",
        )

        self.assertEqual(
            response.status_code,
            200,
        )

        self.assertEqual(
            response.json()["answer"],
            "پاسخ آزمایشی",
        )

    def test_chat_rejects_empty_message(self):
        response = self.client.post(
            reverse("ai_chat:chat"),
            data=json.dumps({
                "message": "",
            }),
            content_type="application/json",
        )

        self.assertEqual(
            response.status_code,
            400,
        )

    def test_chat_rejects_invalid_json(self):
        response = self.client.post(
            reverse("ai_chat:chat"),
            data="{invalid-json",
            content_type="application/json",
        )

        self.assertEqual(
            response.status_code,
            400,
        )

اجرای تست:

python manage.py test

در تست‌های عادی، اتصال به API واقعی را Mock کنید؛ زیرا اتصال واقعی هزینه دارد، به شبکه وابسته است و پاسخ مدل ممکن است ثابت نباشد.

تست کیفیت خروجی مدل

موفق‌بودن وضعیت HTTP به معنی مناسب‌بودن پاسخ نیست.

برای قابلیت تحلیل بازخورد، این ورودی‌ها را آزمایش کنید:

• نظر مثبت
• نظر منفی
• نظر خنثی
• نظر دارای چند موضوع
• متن بسیار کوتاه
• متن فارسی و انگلیسی ترکیبی
• متن دارای غلط املایی
• ورودی نامرتبط
• متن طولانی

معیارهای پذیرش:

• JSON معتبر باشد.
• Category یکی از مقادیر مجاز باشد.
• Sentiment معتبر باشد.
• Priority بین ۱ تا ۵ باشد.
• Summary خالی نباشد.
• زمان پاسخ قابل‌قبول باشد.
• مصرف توکن از سقف تعیین‌شده عبور نکند.

پس از تغییر مدل، پرامپت یا پارامترها، این مجموعه را دوباره اجرا کنید.

ساخت Dockerfile

فایل Dockerfile:

FROM python:3.13-slim

ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

WORKDIR /app

RUN addgroup --system appgroup \
    && adduser --system \
       --ingroup appgroup \
       appuser

COPY requirements.txt .

RUN pip install \
    --no-cache-dir \
    -r requirements.txt

COPY . .

RUN chown -R appuser:appgroup /app

USER appuser

EXPOSE 8000

CMD [
  "uvicorn",
  "config.asgi:application",
  "--host",
  "0.0.0.0",
  "--port",
  "8000"
]

نسخه Python را با نسخه موردنیاز پروژه هماهنگ کنید.

ساخت Image:

docker build \
  -t darvareh-django-ai .

اجرای Container:

docker run --rm \
  -p 8000:8000 \
  -e DARVAREH_API_KEY="YOUR_DARVAREH_API_KEY" \
  -e DARVAREH_MODEL="YOUR_MODEL_ID" \
  -e DARVAREH_BASE_URL="https://api.darvareh.ir/v1" \
  darvareh-django-ai

API Key را داخل Dockerfile یا Image قرار ندهید.

تنظیم Nginx برای Streaming

اگر Django پشت Nginx اجرا می‌شود، ممکن است پاسخ SSE بافر شود.

نمونه تنظیم مسیر Streaming:

location /api/ai/chat/stream/ {
    proxy_pass http://127.0.0.1:8000;

    proxy_http_version 1.1;

    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

    proxy_buffering off;
    proxy_cache off;

    proxy_read_timeout 180s;
    proxy_send_timeout 180s;

    gzip off;
}

ابتدا Streaming را مستقیماً روی پورت Uvicorn آزمایش کنید. اگر مستقیم درست کار می‌کند اما پشت Nginx پاسخ یک‌باره نمایش داده می‌شود، احتمالاً Buffering فعال است.

چک‌لیست آماده‌سازی برای Production

قبل از انتشار سرویس بررسی کنید:

• API Key خارج از Repository قرار دارد.
• Frontend مستقیماً به درواره متصل نمی‌شود.
• ورودی‌ها اعتبارسنجی می‌شوند.
• اندازه بدنه درخواست محدود شده است.
• Endpointها احراز هویت دارند.
• Rate Limiting فعال است.
• مصرف هر کاربر ثبت می‌شود.
• Timeout مشخص شده است.
• Retry فقط برای خطاهای موقت اجرا می‌شود.
• خروجی JSON مدل اعتبارسنجی می‌شود.
• API Key در Log ذخیره نمی‌شود.
• اطلاعات حساس کاربران بدون ضرورت ثبت نمی‌شوند.
• تاریخچه قدیمی خلاصه می‌شود.
• Cache فقط برای داده مناسب فعال است.
• Streaming تحت ASGI اجرا می‌شود.
• Buffering مسیر SSE غیرفعال است.
• Health Check وجود دارد.
• تست‌های واحد و کیفیت نوشته شده‌اند.
• عملیات طولانی وارد Job Queue می‌شوند.
• مدل و پرامپت نسخه‌بندی شده‌اند.
• سقف هزینه روزانه یا ماهانه تعریف شده است.
• رفتار Fallback مشخص شده است.

خطاهای رایج اتصال Django به API هوش مصنوعی

خطای DARVAREH_API_KEY has not been configured

متغیر محیطی تعریف نشده است:

export DARVAREH_API_KEY="YOUR_DARVAREH_API_KEY"

در PowerShell:

$env:DARVAREH_API_KEY="YOUR_DARVAREH_API_KEY"

خطای DARVAREH_MODEL has not been configured

شناسه مدل را تنظیم کنید:

export DARVAREH_MODEL="YOUR_MODEL_ID"

شناسه مدل را از صفحه مدل‌های درواره دریافت کنید.

خطای 401

موارد زیر را بررسی کنید:

• API Key صحیح باشد.
• مقدار کلید فاصله اضافی نداشته باشد.
• Header احراز هویت ارسال شود.
• متغیر محیطی در Process سرور وجود داشته باشد.

ساختار Header:

Authorization: Bearer YOUR_DARVAREH_API_KEY

خطای 404

آدرس کامل Endpoint:

https://api.darvareh.ir/v1/chat/completions

Base URL:

https://api.darvareh.ir/v1

مسیر Client:

chat/completions

خطای CSRF

در یک وب‌سایت Session-based باید CSRF Token را از Frontend ارسال کنید. در APIهای مستقل معمولاً از روش احراز هویت دیگری استفاده می‌شود.

در نمونه آموزشی csrf_exempt فعال شده است، اما برای محصول واقعی باید معماری احراز هویت و CSRF آگاهانه طراحی شود.

Streaming یک‌باره نمایش داده می‌شود

این موارد را بررسی کنید:

• پروژه با ASGI اجرا شود.
• از Uvicorn یا ASGI Server استفاده شود.
• curl -N استفاده شود.
• Nginx Buffering غیرفعال باشد.
• فشرده‌سازی مسیر SSE بررسی شود.
• CDN پاسخ را Buffer نکند.
• Header X-Accel-Buffering: no وجود داشته باشد.

پاسخ JSON قابل Parse نیست

دلایل احتمالی:

• مدل متن توضیحی اطراف JSON نوشته است.
• پاسخ ناقص شده است.
• max_tokens کم است.
• مدل از JSON Mode پشتیبانی نمی‌کند.
• ساختار پرامپت دقیق نیست.

راهکارها:

• Temperature را کاهش دهید.
• پرامپت را دقیق‌تر کنید.
• از مدل مناسب Structured Output استفاده کنید.
• خروجی را اعتبارسنجی کنید.
• در صورت خطا، Retry کنترل‌شده با پرامپت اصلاحی انجام دهید.

خطای Timeout

• مدل سریع‌تری انتخاب کنید.
• طول ورودی را کاهش دهید.
• max_tokens را محدود کنید.
• تاریخچه قدیمی را حذف کنید.
• برای پاسخ طولانی از Streaming استفاده کنید.
• Timeoutهای Uvicorn، Nginx و HTTPX را هماهنگ کنید.

Django یا FastAPI؛ کدام برای هوش مصنوعی بهتر است؟

هر دو فریم‌ورک می‌توانند به API هوش مصنوعی متصل شوند.

اگر پروژه شما از قبل با Django ساخته شده، معمولاً نیازی به مهاجرت به FastAPI نیست. می‌توانید قابلیت هوش مصنوعی را مستقیماً به همان پروژه اضافه کنید.

اگر فقط یک Microservice کوچک و Async برای مدل‌ها می‌سازید، FastAPI نیز انتخاب مناسبی است.

آیا به Django REST Framework نیاز داریم؟

خیر. نمونه این مقاله با Viewهای داخلی Django ساخته شده است.

Django REST Framework یا DRF زمانی مفید است که به این قابلیت‌ها نیاز دارید:

• Serializer
• ViewSet
• Authentication آماده
• Permission
• Throttling
• Browsable API
• Pagination
• Content Negotiation

APIView در DRF پیش از اجرای Handler می‌تواند Authentication، Permission و Throttle را بررسی کند. جزئیات در مستندات رسمی APIView موجود است.

برای مسیر Streaming Async باید سازگاری نسخه DRF و معماری View را با دقت بررسی کنید. استفاده مستقیم از StreamingHttpResponse در یک Async View جنگو، مسیر شفاف‌تری برای SSE است.

چرا از درواره در پروژه Django استفاده کنیم؟

اتصال جداگانه به سرویس‌های مختلف هوش مصنوعی باعث افزایش تعداد API Keyها، تفاوت Payloadها و پیچیدگی نگه‌داری می‌شود.

درواره یک API یکپارچه در اختیار توسعه‌دهندگان قرار می‌دهد. در نتیجه می‌توانید:

• از یک Base URL ثابت استفاده کنید.
• API Key را در Backend نگه دارید.
• مدل متناسب با هر کاربرد را انتخاب کنید.
• بدون بازنویسی معماری، مدل را تغییر دهید.
• از مدل‌های متنی مختلف استفاده کنید.
• هزینه مدل‌ها را در یک مسیر بررسی کنید.
• همان اتصال را در Django، FastAPI و سایر Backendها به کار ببرید.

برای شروع، وارد درواره شوید، API Key بسازید و مدل مناسب را از صفحه مدل‌ها انتخاب کنید.

پرسش‌های متداول

آیا می‌توان با Django برنامه هوش مصنوعی ساخت؟

بله. Django برای ساخت Backend چت‌بات، دستیار سازمانی، ابزار تولید محتوا، تحلیل متن و پردازش اسناد مناسب است.

آیا برای استفاده از هوش مصنوعی در Django به GPU نیاز داریم؟

اگر از API استفاده کنید، خیر. پردازش مدل روی زیرساخت سرویس انجام می‌شود و Django فقط درخواست و پاسخ را مدیریت می‌کند.

آیا Django برای چت‌بات مناسب است؟

بله. Django می‌تواند کاربران، تاریخچه گفتگو، اشتراک، محدودیت مصرف و ارتباط با مدل را مدیریت کند.

آیا Django از Async پشتیبانی می‌کند؟

بله. Django از Async View و زنجیره Async تحت ASGI پشتیبانی می‌کند. برای Streaming و اتصال‌های طولانی بهتر است از ASGI Server استفاده کنید.

آیا می‌توان از Requests به‌جای HTTPX استفاده کرد؟

بله، اما Requests یک Client هم‌زمان است. برای Async View و Streaming غیرمسدودکننده، HTTPX AsyncClient انتخاب مناسب‌تری است.

آیا می‌توان تاریخچه مکالمه را در PostgreSQL ذخیره کرد؟

بله. مدل‌های Django در این مقاله با SQLite، PostgreSQL و سایر پایگاه‌های داده پشتیبانی‌شده قابل‌استفاده‌اند.

آیا API Key را می‌توان در JavaScript قرار داد؟

خیر. API Key باید فقط در Backend جنگو نگه‌داری شود.

آیا خروجی مدل همیشه JSON معتبر است؟

خیر. حتی با JSON Mode باید خروجی را Parse و بر اساس قوانین برنامه اعتبارسنجی کنید.

آیا می‌توان پاسخ مدل را Stream کرد؟

بله. با StreamingHttpResponse، Async Generator و اجرای Django تحت ASGI می‌توانید پاسخ را به‌صورت SSE ارسال کنید.

بهترین مدل برای Django کدام است؟

Django به مدل خاصی وابسته نیست. مدل را بر اساس کیفیت، سرعت، هزینه، Context Window و نوع وظیفه انتخاب کنید. فهرست مدل‌ها در صفحه مدل‌های درواره موجود است.

چگونه هزینه API را کاهش دهیم؟

طول ورودی و خروجی را محدود کنید، تاریخچه را خلاصه کنید، پاسخ‌های تکراری را Cache کنید، مدل مناسب انتخاب کنید و مصرف هر کاربر را ثبت کنید.

آیا می‌توان Django را چندسروری کرد؟

بله. برای این کار بهتر است PostgreSQL، Redis، Object Storage و Queue میان Instanceها مشترک باشند و سرورها پشت Load Balancer قرار گیرند.

آیا برای عملیات طولانی باید از Celery استفاده کنیم؟

برای کارهایی که بیشتر از چرخه معمول درخواست HTTP زمان می‌برند، استفاده از Celery یا یک Job Queue انتخاب مناسب‌تری است.

جمع‌بندی

Django ابزارهای لازم برای ساخت Backend یک محصول هوش مصنوعی را در اختیار توسعه‌دهندگان قرار می‌دهد. سیستم کاربران، ORM، پنل مدیریت، Cache، Middleware و پشتیبانی از ASGI باعث می‌شوند بتوانید قابلیت هوش مصنوعی را در کنار منطق اصلی محصول پیاده‌سازی کنید.

در این آموزش یک پروژه عملی ساختیم که:

• به API درواره متصل می‌شود.
• API Key را خارج از کد نگه می‌دارد.
• پاسخ متنی دریافت می‌کند.
• خلاصه‌سازی انجام می‌دهد.
• بازخورد را به JSON تبدیل می‌کند.
• خروجی مدل را اعتبارسنجی می‌کند.
• پاسخ را به‌صورت Streaming ارائه می‌دهد.
• مکالمات و مصرف توکن را ذخیره می‌کند.
• تحت ASGI اجرا می‌شود.
• با Docker قابل‌استقرار است.
• قابلیت گسترش به Cache، Celery و PostgreSQL را دارد.

برای ساخت اولین سرویس هوش مصنوعی با Django، در درواره ثبت‌نام کنید، API Key بسازید و شناسه مدل موردنظر را از صفحه مدل‌ها انتخاب کنید.

مقالات مرتبط

• API هوش مصنوعی چیست؟ راهنمای کامل توسعه‌دهندگان
• آموزش دریافت API Key هوش مصنوعی
• API سازگار با OpenAI چیست؟
• چگونه API هوش مصنوعی را به نرم‌افزار خود اضافه کنیم؟
• آموزش ساخت چت‌بات با API درواره
• آموزش Streaming API در هوش مصنوعی
• راهنمای Structured Outputs و JSON Schema
• توکن در API هوش مصنوعی چیست؟
• روش‌های کاهش هزینه API هوش مصنوعی
• ساخت API هوش مصنوعی آماده Production
• مانیتورینگ و Observability سرویس‌های هوش مصنوعی
• Fallback در سرویس‌های هوش مصنوعی

برای مطالعه شرایط استفاده و محدودیت‌های مسئولیت، صفحه «سلب مسئولیت» را مشاهده کنید.