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

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

این راهنما برای سازمان‌ها و نمایندگان فروش 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.5",
        "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.5",
    instructions="You are a helpful assistant.",
    input="Write a one-sentence summary of AvalAI.",
)

print(response.output_text)
  • messagesinput
  • پیام سیستمی → instructions یا آیتم developer
  • choices[0].message.contentresponse.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("x-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)
  • messagesinput
  • پیام سیستمی → instructions یا آیتم developer
  • choices[0].message.contentresponse.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('x-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.5",
    instructions="You are a helpful assistant.",
    input="Write a one-sentence summary of AvalAI.",
)

print(response.output_text)
  • messagesinput
  • پیام سیستمی → instructions یا آیتم developer
  • choices[0].message.contentresponse.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 تماس بگیرید.