داشبورد توسعه‌دهنده
پرسش از هوش مصنوعی
پرسش از هوش مصنوعی

راهنمای استفاده سازمانی ​

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

فهرست مطالب ​


نمای کلی ​

سازمان‌هایی که از AvalAI API در مقیاس بزرگ استفاده می‌کنند، نیازمندی‌های منحصر به فردی دارند:

  • حجم بالا: صدها هزار یا میلیون‌ها فراخوانی API روزانه
  • چند مستاجره: چندین بخش، مشتری یا پروژه
  • تخصیص هزینه: پیگیری تفصیلی هزینه به ازای هر واحد تجاری
  • انطباق: مسیرهای حسابرسی و حاکمیت داده
  • عملکرد: حداقل سربار تاخیر برای پیگیری
  • تحلیل‌ها: هوش تجاری و بینش استفاده

User API (/user/v1/) پایه‌ای برای عملیات در مقیاس سازمانی فراهم می‌کند.


موارد استفاده سازمانی ​

1. ارائه‌دهنده سرویس IT داخلی ​

سناریو: شرکت بزرگ که سرویس‌های هوش مصنوعی به چندین واحد تجاری ارائه می‌دهد

نیازمندی‌ها:

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

راه‌حل:

python
# برچسب‌گذاری درخواست‌ها با شناسه بخش
response = requests.post(
    "https://api.avalai.ir/v1/chat/completions",
    headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
    json={
        "model": "gpt-5.6-luna",
        "messages": messages,
        "safety_identifier": f"dept-{department_id}",  # شناسه سفارشی در بدنه
    },
)

# بعدا، فیلتر تراکنش‌ها بر اساس بخش
transactions = get_transactions(safety_identifier=f"dept-{department_id}")
نسخه معادل Responses API

وقتی مدل انتخابی از /v1/responses پشتیبانی می‌کند، این نسخه را کنار مثال Chat Completions استفاده کنید. messages به input منتقل می‌شود و متن نهایی از response.output_text خوانده می‌شود.

python
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["AVALAI_API_KEY"],
    base_url="https://api.avalai.ir/v1",
)

response = client.responses.create(
    model="gpt-5.6-luna",
    instructions="You are a helpful assistant.",
    input="Write a one-sentence summary of AvalAI.",
)

print(response.output_text)
  • messages → input
  • پیام سیستمی → instructions یا آیتم developer
  • choices[0].message.content → response.output_text
  • برای ابزارها و خروجی‌های چندوجهی، response.output را بر اساس type بررسی کنید.

2. ارائه‌دهنده پلتفرم SaaS ​

سناریو: شرکت SaaS که ویژگی‌های هوش مصنوعی را برای هزاران مشتری تعبیه کرده

نیازمندی‌ها:

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

راه‌حل:

python
# پروکسی API اختصاصی مشتری
class CustomerAPIProxy:
    def __init__(self, customer_id):
        self.customer_id = customer_id
        self.api_key = get_api_key_for_customer(customer_id)

    async def chat_completion(self, **kwargs):
        # بررسی سهمیه مشتری
        if not await self.check_quota():
            raise QuotaExceededError()

        # درخواست با پیگیری مشتری
        response = await make_api_request(**kwargs)
        request_id = response.headers.get("avalai-request-id")

        # صف پیگیری هزینه
        await queue_cost_lookup(request_id, self.customer_id)

        return response.json()

نکته: استفاده از OpenAI SDK هنگام استفاده از OpenAI Python SDK به جای درخواست‌های HTTP خام، می‌توانید شناسه درخواست را راحت‌تر دریافت کنید:

python
from openai import OpenAI

client = OpenAI(api_key="your-avalai-api-key", base_url="https://api.avalai.ir/v1")
response = client.chat.completions.create(model="gpt-5.4-mini", messages=messages)

# دسترسی مستقیم به request_id از شیء پاسخ
request_id = response._request_id
نسخه معادل Responses API

وقتی مدل انتخابی از /v1/responses پشتیبانی می‌کند، این نسخه را کنار مثال Chat Completions استفاده کنید. messages به input منتقل می‌شود و متن نهایی از response.output_text خوانده می‌شود.

python
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["AVALAI_API_KEY"],
    base_url="https://api.avalai.ir/v1",
)

response = client.responses.create(
    model="gpt-5.4-mini",
    instructions="You are a helpful assistant.",
    input="Write a one-sentence summary of AvalAI.",
)

print(response.output_text)
  • messages → input
  • پیام سیستمی → instructions یا آیتم developer
  • choices[0].message.content → response.output_text
  • برای ابزارها و خروجی‌های چندوجهی، response.output را بر اساس type بررسی کنید.

این روش توصیه شده هنگام استفاده از OpenAI SDK است زیرا تجزیه هدر را به طور خودکار انجام می‌دهد.

نکته: استفاده از LangChain LangChain به طور مستقیم هدرهای پاسخ HTTP را نمایش نمی‌دهد. از یک کلاینت HTTP سفارشی برای ذخیره آنها استفاده کنید:

python
import httpx
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import BaseCallbackHandler
from contextvars import ContextVar
import os

request_headers: ContextVar[dict] = ContextVar('request_headers', default={})

class HeaderCapturingClient(httpx.Client):
    def send(self, request, **kwargs):
        response = super().send(request, **kwargs)
        request_headers.set(dict(response.headers))
        return response

class HeaderAccessCallback(BaseCallbackHandler):
    def __init__(self):
        self.request_id = None

    def on_llm_end(self, response, **kwargs):
        headers = request_headers.get()
        self.request_id = headers.get('avalai-request-id')

http_client = HeaderCapturingClient()
chat = ChatOpenAI(
    base_url="https://api.avalai.ir/v1",
    api_key=os.getenv("AVALAI_API_KEY"),
    model="gpt-5.4-mini",
    http_client=http_client,
)

callback = HeaderAccessCallback()
response = chat.invoke("سلام", config={"callbacks": [callback]})
request_id = callback.request_id  # اکنون برای پیگیری هزینه در دسترس است

برای مثال‌های غیرهمزمان به راهنمای هدرهای پاسخ مراجعه کنید.

3. نماینده فروش سرویس‌های هوش مصنوعی ​

سناریو: شرکت فروش مجدد دسترسی API هوش مصنوعی با مارک‌آپ

نیازمندی‌ها:

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

راه‌حل: به راهنمای پیگیری هزینه نمایندگان مراجعه کنید

4. سازمان تحقیقاتی ​

سناریو: دانشگاه یا آزمایشگاه تحقیقاتی با چندین پروژه

نیازمندی‌ها:

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

پیگیری هزینه با حجم بالا ​

برای سازمان‌هایی که میلیون‌ها درخواست را پردازش می‌کنند، پیگیری کارآمد هزینه حیاتی است.

استراتژی پردازش دسته‌ای ​

python
import asyncio
from typing import List
from datetime import datetime, timedelta


class EnterpriseCostTracker:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.avalai.ir"
        self.pending_requests = []

    async def track_request(self, request_id: str, metadata: dict):
        """صف کردن درخواست برای پیگیری هزینه"""
        self.pending_requests.append(
            {
                "request_id": request_id,
                "metadata": metadata,
                "queued_at": datetime.now(),
            }
        )

    async def process_batch(self, batch_size: int = 1000):
        """پردازش درخواست‌های معلق در دسته‌ها"""
        while self.pending_requests:
            # تا ۱۰۰۰ درخواست بردارید
            batch = self.pending_requests[:batch_size]
            request_ids = [r["request_id"] for r in batch]

            # جستجوی هزینه‌ها در دسته
            response = await self.lookup_costs(request_ids)

            # پردازش نتایج
            for transaction in response["transactions"]:
                request_id = transaction["id"]
                metadata = next(
                    r["metadata"] for r in batch if r["request_id"] == request_id
                )

                await self.record_cost(transaction, metadata)

            # حذف درخواست‌های پردازش شده
            self.pending_requests = self.pending_requests[batch_size:]

    async def lookup_costs(self, request_ids: List[str]):
        """جستجوی هزینه‌ها برای چندین درخواست"""
        response = await asyncio.get_event_loop().run_in_executor(
            None,
            lambda: requests.post(
                f"{self.base_url}/user/v1/transactions/lookup",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json",
                },
                json={"transaction_ids": request_ids},
            ),
        )
        return response.json()

    async def record_cost(self, transaction: dict, metadata: dict):
        """ثبت هزینه در پایگاه داده"""
        cost_usd = float(transaction["cost"]["unit"])
        cost_irt = float(transaction["cost"]["paid_irt"]) + float(
            transaction["cost"]["paid_grant_irt"]
        )

        # ذخیره در پایگاه داده با metadata
        await db.insert_cost_record(
            {
                "request_id": transaction["id"],
                "customer_id": metadata.get("customer_id"),
                "department_id": metadata.get("department_id"),
                "project_id": metadata.get("project_id"),
                "cost_usd": cost_usd,
                "cost_irt": cost_irt,
                "model": transaction["model"],
                "tokens": transaction["tokens"],
                "timestamp": transaction["requested_at"],
            }
        )


# استفاده
tracker = EnterpriseCostTracker(api_key)

# پیگیری درخواست‌ها به محض ورود
await tracker.track_request(
    request_id,
    {"customer_id": "cust_123", "department_id": "eng", "project_id": "proj_456"},
)

# پردازش در پس‌زمینه هر دقیقه
while True:
    await tracker.process_batch(batch_size=1000)
    await asyncio.sleep(60)

معماری چند مستاجره ​

الگوی معماری ​

┌─────────────────────────────────────────────────────┐
│                   Load Balancer                      │
└──────────────────────┬──────────────────────────────┘
                       │
        ┌──────────────┼──────────────┐
        │              │              │
   ┌────▼────┐    ┌───▼────┐    ┌───▼────┐
   │  API    │    │  API   │    │  API   │
   │ Server 1│    │Server 2│    │Server 3│
   └────┬────┘    └───┬────┘    └───┬────┘
        │             │              │
        └──────────────┼──────────────┘
                       │
        ┌──────────────▼──────────────┐
        │    صف پیگیری درخواست         │
        │        (Redis/Kafka)         │
        └──────────────┬──────────────┘
                       │
        ┌──────────────▼──────────────┐
        │     پردازش هزینه             │
        │      Workers (Async)         │
        └──────────────┬──────────────┘
                       │
        ┌──────────────▼──────────────┐
        │      پایگاه داده              │
        │   (PostgreSQL/MongoDB)       │
        └──────────────────────────────┘

مثال طرحواره پایگاه داده ​

sql
-- جدول مشتریان/مستاجرها
CREATE TABLE customers (
    id UUID PRIMARY KEY,
    name VARCHAR(255),
    api_key_id INTEGER,
    pricing_tier VARCHAR(50),
    created_at TIMESTAMP
);

-- جدول پیگیری درخواست
CREATE TABLE api_requests (
    request_id UUID PRIMARY KEY,
    customer_id UUID REFERENCES customers(id),
    model VARCHAR(100),
    requested_at TIMESTAMP,
    cost_usd DECIMAL(12, 8),
    cost_irt DECIMAL(12, 2),
    tokens_total INTEGER,
    tokens_prompt INTEGER,
    tokens_completion INTEGER,
    status VARCHAR(50),
    metadata JSONB
);

-- ایندکس‌ها برای عملکرد
CREATE INDEX idx_requests_customer ON api_requests(customer_id);
CREATE INDEX idx_requests_timestamp ON api_requests(requested_at);
CREATE INDEX idx_requests_status ON api_requests(status);

-- نمای خلاصه استفاده روزانه
CREATE MATERIALIZED VIEW daily_usage_summary AS
SELECT 
    customer_id,
    DATE(requested_at) as usage_date,
    COUNT(*) as request_count,
    SUM(cost_usd) as total_cost_usd,
    SUM(cost_irt) as total_cost_irt,
    SUM(tokens_total) as total_tokens
FROM api_requests
WHERE status = 'completed'
GROUP BY customer_id, DATE(requested_at);

-- به‌روزرسانی دوره‌ای نمای materialized
REFRESH MATERIALIZED VIEW CONCURRENTLY daily_usage_summary;

تحلیل‌های پیشرفته ​

داشبورد استفاده در زمان واقعی ​

python
from fastapi import FastAPI
from datetime import datetime, timedelta

app = FastAPI()


@app.get("/analytics/realtime/{customer_id}")
async def get_realtime_analytics(customer_id: str):
    """تحلیل استفاده در زمان واقعی برای مشتری"""
    now = datetime.now()

    # آمار ساعت گذشته
    hour_stats = await db.query(
        """
        SELECT 
            COUNT(*) as requests,
            SUM(cost_usd) as cost,
            SUM(tokens_total) as tokens,
            AVG(tokens_total) as avg_tokens_per_request
        FROM api_requests
        WHERE customer_id = $1
        AND requested_at >= $2
    """,
        customer_id,
        now - timedelta(hours=1),
    )

    # توزیع مدل
    model_dist = await db.query(
        """
        SELECT 
            model,
            COUNT(*) as count,
            SUM(cost_usd) as cost
        FROM api_requests
        WHERE customer_id = $1
        AND requested_at >= $2
        GROUP BY model
        ORDER BY count DESC
    """,
        customer_id,
        now - timedelta(hours=24),
    )

    # روند هزینه (۷ روز گذشته)
    cost_trend = await db.query(
        """
        SELECT 
            DATE(requested_at) as date,
            SUM(cost_usd) as daily_cost
        FROM api_requests
        WHERE customer_id = $1
        AND requested_at >= $2
        GROUP BY DATE(requested_at)
        ORDER BY date
    """,
        customer_id,
        now - timedelta(days=7),
    )

    return {
        "customer_id": customer_id,
        "last_hour": hour_stats,
        "model_distribution": model_dist,
        "cost_trend": cost_trend,
    }

بهترین شیوه‌های امنیتی ​

مدیریت کلید API ​

python
# کلیدهای API جداگانه برای هر مشتری/مستاجر
class APIKeyManager:
    def __init__(self):
        self.key_store = {}  # از مدیریت secrets مناسب استفاده کنید

    def create_customer_key(self, customer_id: str) -> str:
        """ایجاد کلید API اختصاصی برای مشتری"""
        # از کلید اصلی AvalAI خود به صورت داخلی استفاده کنید
        # ایجاد رکورد پیگیری
        key_id = self.register_key(customer_id)
        return f"customer_{customer_id}_key_{key_id}"

    def get_avalai_key(self, customer_key: str) -> str:
        """نگاشت کلید مشتری به کلید اصلی AvalAI"""
        # اعتبارسنجی کلید مشتری
        if not self.validate_key(customer_key):
            raise UnauthorizedError()

        # بازگشت کلید اصلی AvalAI
        return os.getenv("AVALAI_MASTER_API_KEY")

محدودیت نرخ ​

python
from redis import Redis
from datetime import datetime, timedelta


class EnterpriseRateLimiter:
    def __init__(self, redis_client: Redis):
        self.redis = redis_client

    async def check_limit(self, customer_id: str, limit_per_minute: int = 1000) -> bool:
        """بررسی اینکه مشتری در محدودیت‌های نرخ است"""
        key = f"rate_limit:{customer_id}:{datetime.now().strftime('%Y%m%d%H%M')}"

        count = await self.redis.incr(key)
        if count == 1:
            await self.redis.expire(key, 60)

        return count <= limit_per_minute

بهینه‌سازی عملکرد ​

استراتژی کش ​

python
from functools import lru_cache
import redis


class CostCache:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.cache_ttl = 3600  # ۱ ساعت

    async def get_cached_cost(self, request_id: str) -> dict:
        """دریافت هزینه از کش در صورت وجود"""
        cached = await self.redis.get(f"cost:{request_id}")
        if cached:
            return json.loads(cached)
        return None

    async def cache_cost(self, request_id: str, cost_data: dict):
        """کش کردن داده‌های هزینه"""
        await self.redis.setex(
            f"cost:{request_id}", self.cache_ttl, json.dumps(cost_data)
        )

Connection Pooling ​

python
import aiohttp
from aiohttp import TCPConnector

# استفاده مجدد از اتصالات برای عملکرد بهتر
connector = TCPConnector(limit=100, limit_per_host=30)
session = aiohttp.ClientSession(connector=connector)


async def make_api_request(**kwargs):
    """درخواست API با connection pooling"""
    async with session.post(
        "https://api.avalai.ir/v1/chat/completions", **kwargs
    ) as response:
        return await response.json()
نسخه معادل Responses API

وقتی مدل انتخابی از /v1/responses پشتیبانی می‌کند، این نسخه را کنار مثال Chat Completions استفاده کنید. messages به input منتقل می‌شود و متن نهایی از response.output_text خوانده می‌شود.

python
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["AVALAI_API_KEY"],
    base_url="https://api.avalai.ir/v1",
)

response = client.responses.create(
    model="gpt-5.6-luna",
    instructions="You are a helpful assistant.",
    input="Write a one-sentence summary of AvalAI.",
)

print(response.output_text)
  • messages → input
  • پیام سیستمی → instructions یا آیتم developer
  • choices[0].message.content → response.output_text
  • برای ابزارها و خروجی‌های چندوجهی، response.output را بر اساس type بررسی کنید.

پیگیری تولید ویدیو ​

برای سازمان‌هایی که از APIهای تولید ویدیو (Sora، Veo، Runway) استفاده می‌کنند، پیگیری با پشتیبانی داخلی request_id و safety_identifier در بدنه پاسخ ساده‌تر شده است.

فیلدهای پاسخ Video API ​

برخلاف chat completions استاندارد که request_id فقط در هدرهای پاسخ موجود است، پاسخ‌های Video API فیلدهای پیگیری را مستقیما در بدنه JSON شامل می‌شوند:

python
import requests

# ایجاد ویدیو با پیگیری سازمانی
response = requests.post(
    "https://api.avalai.ir/v1/videos",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={
        "model": "sora-2",
        "prompt": "ویدیوی نمایش محصول",
        "size": "1280x720",
        "seconds": "4",
        "safety_identifier": f"dept-{department_id}",  # شناسه داخلی شما
    },
)

video = response.json()

# دسترسی به فیلدهای پیگیری مستقیما از بدنه پاسخ
print(f"شناسه ویدیو: {video['id']}")
print(f"شناسه درخواست: {video['request_id']}")  # UUID v7 برای پیگیری هزینه
print(f"شناسه ایمنی: {video.get('safety_identifier')}")

فیلتر کردن ویدیوها بر اساس شناسه ​

سازمان‌ها می‌توانند لیست ویدیوها را بر اساس safety_identifier یا request_id فیلتر کنند:

python
# دریافت تمام ویدیوهای یک بخش خاص
department_videos = requests.get(
    "https://api.avalai.ir/v1/videos",
    headers={"Authorization": f"Bearer {API_KEY}"},
    params={"safety_identifier": f"dept-{department_id}"},
).json()

# دریافت یک ویدیوی خاص بر اساس request_id
specific_video = requests.get(
    "https://api.avalai.ir/v1/videos",
    headers={"Authorization": f"Bearer {API_KEY}"},
    params={"request_id": "019b47a0-ece8-75b2-8a4c-40fcf4b49479"},
).json()

معماری پیگیری هزینه ویدیو ​

برای سازمان‌ها با سرویس‌های پردازش ویدیو همزمان:

python
class VideoTrackingService:
    """
    پیگیری تولید ویدیو در چندین سرویس با استفاده از safety_identifier.

    مورد استفاده: تیم بازاریابی درخواست تولید ویدیو می‌دهد و چندین سرویس
    (صورتحساب، تحلیل‌ها، اعلان) نیاز به استعلام همان ویدیو دارند.
    """

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.avalai.ir"

    async def create_tracked_video(
        self, prompt: str, department_id: str, project_id: str
    ) -> dict:
        """ایجاد ویدیو با شناسه‌های پیگیری سازمانی"""
        # ایجاد safety_identifier ترکیبی برای پیگیری چند بعدی
        safety_id = f"dept_{department_id}_proj_{project_id}"

        response = requests.post(
            f"{self.base_url}/v1/videos",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={
                "model": "sora-2-pro",
                "prompt": prompt,
                "size": "1792x1024",
                "seconds": "8",
                "safety_identifier": safety_id,
            },
        )

        video = response.json()

        # ذخیره request_id برای پیگیری هزینه
        await self.queue_cost_lookup(
            request_id=video["request_id"],
            video_id=video["id"],
            department_id=department_id,
            project_id=project_id,
        )

        return video

    async def get_department_videos(self, department_id: str) -> list:
        """دریافت تمام ویدیوهای یک بخش با استفاده از پیشوند safety_identifier"""
        # فیلتر بر اساس الگوی safety_identifier
        response = requests.get(
            f"{self.base_url}/v1/videos",
            headers={"Authorization": f"Bearer {self.api_key}"},
            params={"safety_identifier": f"dept_{department_id}"},
        )

        return response.json()["data"]

    async def queue_cost_lookup(self, request_id: str, **metadata):
        """صف پیگیری هزینه ویدیو با استفاده از request_id"""
        # استفاده از User API برای پیگیری هزینه‌ها
        cost_response = requests.post(
            f"{self.base_url}/user/v1/transactions/lookup",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={"transaction_ids": [request_id]},
        )

        if cost_response.json()["transactions"]:
            transaction = cost_response.json()["transactions"][0]
            await self.record_video_cost(transaction, metadata)

طرح پایگاه داده برای پیگیری ویدیو ​

sql
-- جدول پیگیری درخواست‌های ویدیو
CREATE TABLE video_requests (
    video_id VARCHAR(64) PRIMARY KEY,
    request_id UUID UNIQUE NOT NULL,
    safety_identifier VARCHAR(256),
    customer_id UUID REFERENCES customers(id),
    department_id VARCHAR(50),
    project_id VARCHAR(50),
    model VARCHAR(100),
    prompt TEXT,
    size VARCHAR(20),
    seconds INTEGER,
    status VARCHAR(20),
    requested_at TIMESTAMP,
    completed_at TIMESTAMP,
    cost_usd DECIMAL(12, 8),
    cost_irt DECIMAL(12, 2)
);

-- ایندکس‌ها برای استعلام‌های سازمانی
CREATE INDEX idx_video_safety_id ON video_requests(safety_identifier);
CREATE INDEX idx_video_request_id ON video_requests(request_id);
CREATE INDEX idx_video_customer ON video_requests(customer_id);
CREATE INDEX idx_video_department ON video_requests(department_id);
CREATE INDEX idx_video_status ON video_requests(status);

منابع مرتبط ​


برای پشتیبانی سازمانی: برای مدیریت حساب اختصاصی و قیمت‌گذاری سازمانی با support@avalai.ir تماس بگیرید.