راهنمای استفاده سازمانی
این راهنما برای سازمانها و نمایندگان فروش API در مقیاس بزرگ طراحی شده که به پیگیری استفاده پیشرفته، مدیریت هزینه و بینش عملیاتی برای استقرارهای AvalAI API با حجم بالا نیاز دارند.
فهرست مطالب
- نمای کلی
- موارد استفاده سازمانی
- پیگیری هزینه با حجم بالا
- معماری چند مستاجره
- تحلیلهای پیشرفته
- بهترین شیوههای امنیتی
- بهینهسازی عملکرد
نمای کلی
سازمانهایی که از AvalAI API در مقیاس بزرگ استفاده میکنند، نیازمندیهای منحصر به فردی دارند:
- حجم بالا: صدها هزار یا میلیونها فراخوانی API روزانه
- چند مستاجره: چندین بخش، مشتری یا پروژه
- تخصیص هزینه: پیگیری تفصیلی هزینه به ازای هر واحد تجاری
- انطباق: مسیرهای حسابرسی و حاکمیت داده
- عملکرد: حداقل سربار تاخیر برای پیگیری
- تحلیلها: هوش تجاری و بینش استفاده
User API (/user/v1/) پایهای برای عملیات در مقیاس سازمانی فراهم میکند.
موارد استفاده سازمانی
1. ارائهدهنده سرویس IT داخلی
سناریو: شرکت بزرگ که سرویسهای هوش مصنوعی به چندین واحد تجاری ارائه میدهد
نیازمندیها:
- پیگیری استفاده به ازای هر بخش
- گزارشدهی chargeback/showback
- هشدارها و کنترلهای بودجه
- مسیرهای حسابرسی انطباق
راهحل:
# برچسبگذاری درخواستها با شناسه بخش
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 خوانده میشود.
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)messages→input- پیام سیستمی →
instructionsیا آیتمdeveloper choices[0].message.content→response.output_text- برای ابزارها و خروجیهای چندوجهی،
response.outputرا بر اساسtypeبررسی کنید.
2. ارائهدهنده پلتفرم SaaS
سناریو: شرکت SaaS که ویژگیهای هوش مصنوعی را برای هزاران مشتری تعبیه کرده
نیازمندیها:
- پیگیری استفاده به ازای هر مشتری
- محاسبه هزینه در زمان واقعی
- صورتحساب مبتنی بر استفاده
- محدودیت نرخ به ازای هر مشتری
راهحل:
# پروکسی 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 خام، میتوانید شناسه درخواست را راحتتر دریافت کنید:
pythonfrom 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 خوانده میشود.
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 سفارشی برای ضبط آنها استفاده کنید:
pythonimport 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. سازمان تحقیقاتی
سناریو: دانشگاه یا آزمایشگاه تحقیقاتی با چندین پروژه
نیازمندیها:
- تخصیص هزینه مبتنی بر گرنت
- گزارشهای استفاده به ازای هر پروژه
- کنترلهای دسترسی محقق
- مسیرهای حسابرسی برای انتشارات
پیگیری هزینه با حجم بالا
برای سازمانهایی که میلیونها درخواست را پردازش میکنند، پیگیری کارآمد هزینه حیاتی است.
استراتژی پردازش دستهای
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) │
└──────────────────────────────┘مثال طرحواره پایگاه داده
-- جدول مشتریان/مستاجرها
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;تحلیلهای پیشرفته
داشبورد استفاده در زمان واقعی
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
# کلیدهای 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")محدودیت نرخ
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بهینهسازی عملکرد
استراتژی کش
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
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 خوانده میشود.
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)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 شامل میشوند:
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 فیلتر کنند:
# دریافت تمام ویدیوهای یک بخش خاص
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()معماری پیگیری هزینه ویدیو
برای سازمانها با سرویسهای پردازش ویدیو همزمان:
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)طرح پایگاه داده برای پیگیری ویدیو
-- جدول پیگیری درخواستهای ویدیو
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);منابع مرتبط
- مرجع User API - مستندات کامل API
- مرجع Video API - مستندات API تولید ویدیو
- راهنمای پیگیری هزینه نمایندگان - گردش کار تفصیلی پیگیری هزینه
- راهنمای هدرهای پاسخ - درک پیگیری درخواست
- راهنمای محدودیتهای نرخ - سطوح محدودیت نرخ و بهترین شیوهها
برای پشتیبانی سازمانی: برای مدیریت حساب اختصاصی و قیمتگذاری سازمانی با support@avalai.ir تماس بگیرید.