چگونه API AvalAI از دسترسی مستقیم به OpenAI سریع‌تر است؟

یک ادعای جسورانه، یا شاید هم نه؟

بیایید یک راست به سراغ اصل مطلب برویم: فراخوانی API در AvalAI از دسترسی مستقیم به OpenAI سریع‌تر است. نه، اشتباه تایپی نیست. مشکل فنی هم نداریم. و نه، از GPT برای نوشتن این مقاله استفاده نکرده‌ایم (خب، شاید فقط کمی).

اعداد چه می‌گویند:

• اروپا: ۳۹٪ سریع‌تر (۰.۴۳۵ ثانیه در مقابل ۰.۷۱۷ ثانیه)

• خاورمیانه: ۴۴٪ سریع‌تر (۰.۶۶۸ ثانیه در مقابل ۱.۰۴۸ ثانیه)

• توان عملیاتی: ۷۷٪ بیشتر در هر دو منطقه

می‌دانیم به چه فکر می‌کنید: «یک واسط API دیگر که ادعا می‌کند از منبع اصلی سریع‌تر است.» ما هم وقتی برای اولین بار این اعداد را دیدیم، همین فکر را کردیم.

بخش «این با عقل جور در نمی‌آید»

صادق باشیم: طبق قوانین فیزیک، افزودن یک لایه میانی باید باعث کندی شود، نه سرعت. هر توسعه‌دهنده‌ای که زمانی یک Reverse Proxy راه‌اندازی کرده باشد، این را می‌داند.

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

نتایج بنچمارک اروپا (مهر ۱۴۰۴)

┌─────────────────────┬──────────┬─────────┐
│ معیار              │ OpenAI   │ AvalAI  │
├─────────────────────┼──────────┼─────────┤
│ میانه TTFB (ثانیه) │ 0.393    │ 0.685   │
└─────────────────────┴──────────┴─────────┘

Plain text

نتایج بنچمارک خاورمیانه (مهر ۱۴۰۴)

┌─────────────────────┬──────────┬─────────┐
│ معیار              │ OpenAI   │ AvalAI  │
├─────────────────────┼──────────┼─────────┤
│ میانه TTFB (ثانیه) │ 0.668    │ 1.048   │
└─────────────────────┴──────────┴─────────┘

Plain text

خب، حالا که توجه شما را جلب کردیم، بیایید توضیح دهیم این اتفاق چگونه ممکن شده است.

شیرجه‌ای در اعماق فنی (بخش خسته‌کننده اما حیاتی)

راز اول: استخر اتصالات (اصلی که همه می‌دانند، اما کمتر کسی اجرا می‌کند)

تصور کنید هر بار که می‌خواهید با OpenAI ارتباط برقرار کنید، باید این مراحل را طی کنید:

• TLS Handshake (حدود ۱۰۰-۲۰۰ میلی‌ثانیه سربار)

• برقراری اتصال TCP

• احراز هویت

• ارسال درخواست

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

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

# توسعه‌دهنده:
for request in requests:
    connection = establish_connection()  # ۱۰۰ms سربار
    response = send_request(connection)
    close_connection()

# AvalAI:
connection_pool = maintain_warm_connections()  # همیشه آماده
for request in requests:
    response = send_request(connection_pool.get())  # بدون سربار

Python

راز دوم: حجم بالا = یک مزیت ذاتی

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

اما یک توسعه‌دهنده مستقل؟ او باید هر بار از ابتدا شروع کند.

راز سوم: بهینه‌سازی‌های طاقت‌فرسا (که واقعا کار می‌کنند)

در سه ماه گذشته، تیم ما:

• هر میلی‌ثانیه سربار (Overhead) را شناسایی و حذف کرد.

• لایه شبکه را بهینه‌سازی کرد.

• مسیریابی را تنظیم کرد.

• کدی را که ۵ میلی‌ثانیه بیشتر زمان می‌برد، بهسازی (Refactor) کرد.

می‌دانیم، خیلی هیجان‌انگیز به نظر نمی‌رسد. اما همین کارهای به ظاهر کوچک، تفاوت‌های بزرگ را رقم می‌زنند.

پیچش داستان: چرا تجمیع‌کننده‌ها (قاعدتا) باید کندتر باشند؟

بیایید بررسی کنیم: چرا همه تصور می‌کنند استفاده از یک واسطه باید کندتر باشد؟

باور رایج:

• یک گام (Hop) اضافی = تاخیر بیشتر

• یک لایه پردازش اضافی = سربار (Overhead)

• منطقی به نظر می‌رسد، درست است؟

واقعیت فنی:

• اگر تجمیع‌کننده شما استخر اتصالات داشته باشد و شما نداشته باشید ← تجمیع‌کننده سریع‌تر است.

• اگر تجمیع‌کننده شما در همان دیتاسنتر ارائه‌دهنده باشد ← تجمیع‌کننده سریع‌تر است.

• اگر تجمیع‌کننده شما روزانه میلیون‌ها درخواست را پردازش کند ← تجمیع‌کننده بهینه‌تر است.

و اما بخش طعنه‌آمیز ماجرا: بیشتر توسعه‌دهندگان به درستی استخر اتصالات را پیاده‌سازی نمی‌کنند. چرا؟ چون کار خسته‌کننده‌ای است و بسته توسعه نرم‌افزار (SDK) پیش‌فرض OpenAI آنقدرها هم که فکر می‌کنید، هوشمند نیست.

چالش «خودتان امتحان کنید»

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

import os
import requests
import time
import numpy as np
import matplotlib.pyplot as plt
import json
from tabulate import tabulate
from tqdm import tqdm
from datetime import datetime
import seaborn as sns


def test_api_performance(
    api_name, api_url, api_key, model, num_requests=10, prompt="Say hi"
):
    """Test API performance and collect comprehensive metrics"""
    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"}
    data = {"model": model, "messages": [{"role": "user", "content": prompt}]}

    ttfb_times = []
    total_times = []
    token_counts = []
    tokens_per_second = []
    errors = 0

    print(f"Testing {api_name} API with {num_requests} requests...")
    for _ in tqdm(range(num_requests)):
        try:
            start_time = time.time()
            response = requests.post(
                api_url, headers=headers, json=data, timeout=(10, 30)
            )
            response_time = time.time()

            # Process the response
            response_json = response.json()
            end_time = time.time()

            # Calculate metrics
            ttfb = response_time - start_time
            total_time = end_time - start_time

            # Try to get token count if available
            try:
                usage = response_json.get("usage", {})
                total_tokens = usage.get("total_tokens", 0)
                completion_tokens = usage.get("completion_tokens", 0)
                token_counts.append(total_tokens)

                # Calculate tokens per second (using completion tokens)
                if completion_tokens > 0 and total_time > 0:
                    tokens_per_second.append(completion_tokens / total_time)
                else:
                    tokens_per_second.append(0)
            except Exception as e:
                token_counts.append(0)
                tokens_per_second.append(0)

            ttfb_times.append(ttfb)
            total_times.append(total_time)

            # Add a small delay to avoid rate limiting
            time.sleep(0.5)

        except Exception as e:
            print(f"Error on request: {e}")
            errors += 1

    return {
        "name": api_name,
        "url": api_url,
        "model": model,
        "average_ttfb": np.mean(ttfb_times) if ttfb_times else None,
        "median_ttfb": np.median(ttfb_times) if ttfb_times else None,
        "p95_ttfb": np.percentile(ttfb_times, 95) if ttfb_times else None,
        "average_total": np.mean(total_times) if total_times else None,
        "median_total": np.median(total_times) if total_times else None,
        "p95_total": np.percentile(total_times, 95) if total_times else None,
        "ttfb_times": ttfb_times,
        "total_times": total_times,
        "token_counts": token_counts,
        "avg_tokens": (
            np.mean(token_counts) if token_counts and any(token_counts) else None
        ),
        "tokens_per_second": tokens_per_second,
        "avg_tokens_per_second": (
            np.mean([t for t in tokens_per_second if t > 0])
            if tokens_per_second and any(tokens_per_second)
            else None
        ),
        "median_tokens_per_second": (
            np.median([t for t in tokens_per_second if t > 0])
            if tokens_per_second and any(tokens_per_second)
            else None
        ),
        "success_rate": (
            len(ttfb_times) / (len(ttfb_times) + errors)
            if (len(ttfb_times) + errors) > 0
            else 0
        ),
        "error_count": errors,
    }


def print_comparison_table(results_avalai, results_openai):
    """Print comparison table between two APIs"""
    headers = [
        "Metric",
        f"AvalAI ({results_avalai['model']})",
        f"OpenAI ({results_openai['model']})",
    ]

    data = [
        [
            "Average TTFB (s)",
            f"{results_avalai['average_ttfb']:.3f}",
            f"{results_openai['average_ttfb']:.3f}",
        ],
        [
            "Median TTFB (s)",
            f"{results_avalai['median_ttfb']:.3f}",
            f"{results_openai['median_ttfb']:.3f}",
        ],
        [
            "95th Percentile TTFB (s)",
            f"{results_avalai['p95_ttfb']:.3f}",
            f"{results_openai['p95_ttfb']:.3f}",
        ],
        [
            "Average Total Time (s)",
            f"{results_avalai['average_total']:.3f}",
            f"{results_openai['average_total']:.3f}",
        ],
        [
            "Median Total Time (s)",
            f"{results_avalai['median_total']:.3f}",
            f"{results_openai['median_total']:.3f}",
        ],
        [
            "95th Percentile Total (s)",
            f"{results_avalai['p95_total']:.3f}",
            f"{results_openai['p95_total']:.3f}",
        ],
        [
            "Success Rate",
            f"{results_avalai['success_rate']:.2%}",
            f"{results_openai['success_rate']:.2%}",
        ],
    ]

    # Add token metrics if available
    if (
        results_avalai["avg_tokens"] is not None
        and results_openai["avg_tokens"] is not None
    ):
        data.append(
            [
                "Avg Tokens per Response",
                f"{results_avalai['avg_tokens']:.1f}",
                f"{results_openai['avg_tokens']:.1f}",
            ]
        )

    # Add tokens per second metrics if available
    if (
        results_avalai["avg_tokens_per_second"] is not None
        and results_openai["avg_tokens_per_second"] is not None
    ):
        data.append(
            [
                "Avg Tokens per Second",
                f"{results_avalai['avg_tokens_per_second']:.1f}",
                f"{results_openai['avg_tokens_per_second']:.1f}",
            ]
        )
        data.append(
            [
                "Median Tokens per Second",
                f"{results_avalai['median_tokens_per_second']:.1f}",
                f"{results_openai['median_tokens_per_second']:.1f}",
            ]
        )

    print("\nAPI Performance Comparison:")
    print(tabulate(data, headers=headers, tablefmt="grid"))


def plot_comparison(results_avalai, results_openai, output_file=None):
    """Create improved visualization plots for API comparison"""
    # Set the style
    sns.set(style="whitegrid")

    # Create figure with subplots - adding a third subplot for tokens per second
    fig, axes = plt.subplots(3, 1, figsize=(12, 15))

    # Define metrics to plot
    metrics = [
        ("ttfb_times", "Time to First Byte (s)"),
        ("total_times", "Total Request Time (s)"),
        ("tokens_per_second", "Tokens per Second"),
    ]

    # Define colors for each API
    colors = {"AvalAI": "#3498db", "OpenAI": "#2ecc71"}

    for i, (metric, title) in enumerate(metrics):
        # Create violin plots with individual points
        ax = axes[i]

        # Prepare data for plotting
        data_to_plot = []
        labels = []

        for result, label in [(results_avalai, "AvalAI"), (results_openai, "OpenAI")]:
            # Filter out zeros for tokens per second
            if metric == "tokens_per_second":
                data_to_plot.append([t for t in result[metric] if t > 0])
            else:
                data_to_plot.append(result[metric])
            labels.append(f"{label}\n({result['model']})")

        # Create violin plot
        parts = ax.violinplot(data_to_plot, showmeans=True, showmedians=True)

        # Customize violin plots
        for pc, color_key in zip(parts["bodies"], colors.keys()):
            pc.set_facecolor(colors[color_key])
            pc.set_alpha(0.7)

        # Add boxplot inside violin
        bp = ax.boxplot(
            data_to_plot,
            positions=range(1, len(data_to_plot) + 1),
            widths=0.15,
            patch_artist=True,
            showfliers=False,
        )

        # Customize boxplots
        for box, color_key in zip(bp["boxes"], colors.keys()):
            box.set(color="black", linewidth=1.5)
            box.set(facecolor="white")

        # Add scatter points with jitter
        for j, data in enumerate(
            [
                (
                    results_avalai[metric]
                    if metric != "tokens_per_second"
                    else [t for t in results_avalai[metric] if t > 0]
                ),
                (
                    results_openai[metric]
                    if metric != "tokens_per_second"
                    else [t for t in results_openai[metric] if t > 0]
                ),
            ]
        ):
            # Add jitter to x position
            x = np.random.normal(j + 1, 0.05, size=len(data))
            ax.scatter(
                x,
                data,
                alpha=0.4,
                s=20,
                color=list(colors.values())[j],
                edgecolor="white",
                linewidth=0.5,
            )

        # Set labels and title
        ax.set_title(title, fontsize=14, fontweight="bold")
        if metric == "tokens_per_second":
            ax.set_ylabel("Tokens/second", fontsize=12)
        else:
            ax.set_ylabel("Time (seconds)", fontsize=12)
        ax.set_xticks(range(1, len(labels) + 1))
        ax.set_xticklabels(labels, fontsize=12)

        # Add horizontal grid lines
        ax.yaxis.grid(True, linestyle="--", alpha=0.7)

        # Add stats as text
        for j, (result, label) in enumerate(
            [(results_avalai, "AvalAI"), (results_openai, "OpenAI")]
        ):
            if metric == "tokens_per_second":
                if result["avg_tokens_per_second"] is not None:
                    stats = (
                        f"Mean: {result['avg_tokens_per_second']:.1f}\n"
                        f"Median: {result['median_tokens_per_second']:.1f}"
                    )
                    max_val = (
                        max([t for t in result[metric] if t > 0])
                        if any(t > 0 for t in result[metric])
                        else 0
                    )
                    ax.annotate(
                        stats,
                        xy=(j + 1, max_val * 1.05),
                        ha="center",
                        va="bottom",
                        fontsize=10,
                        bbox=dict(boxstyle="round,pad=0.5", fc="white", alpha=0.7),
                    )
            else:
                stats = (
                    f"Mean: {result[f'average_{metric.split("_")[0]}']:.3f}s\n"
                    f"Median: {result[f'median_{metric.split("_")[0]}']:.3f}s\n"
                    f"95th: {result[f'p95_{metric.split("_")[0]}']:.3f}s"
                )
                ax.annotate(
                    stats,
                    xy=(j + 1, result[f'p95_{metric.split("_")[0]}'] * 1.05),
                    ha="center",
                    va="bottom",
                    fontsize=10,
                    bbox=dict(boxstyle="round,pad=0.5", fc="white", alpha=0.7),
                )

    # Add title and timestamp
    plt.suptitle(
        f"API Performance Comparison: AvalAI vs OpenAI", fontsize=16, fontweight="bold"
    )
    plt.figtext(
        0.5,
        0.01,
        f'Generated on {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}',
        ha="center",
        fontsize=10,
    )

    plt.tight_layout(rect=[0, 0.03, 1, 0.97])

    if output_file:
        plt.savefig(output_file, dpi=300, bbox_inches="tight")
        print(f"Plot saved to {output_file}")
    else:
        plt.show()


def save_results(results_avalai, results_openai, filename):
    """Save results to JSON file"""
    # Convert numpy arrays to lists for JSON serialization
    results_avalai_copy = results_avalai.copy()
    results_openai_copy = results_openai.copy()

    for key in ["ttfb_times", "total_times", "token_counts"]:
        if key in results_avalai_copy:
            results_avalai_copy[key] = [float(x) for x in results_avalai_copy[key]]
        if key in results_openai_copy:
            results_openai_copy[key] = [float(x) for x in results_openai_copy[key]]

    data = {
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "results": {"avalai": results_avalai_copy, "openai": results_openai_copy},
    }

    with open(filename, "w") as f:
        json.dump(data, f, indent=2)

    print(f"Results saved to {filename}")


def main():
    # API configuration
    model_name = "gpt-4o-mini"

    url_avalai = "https://api.avalai.ir/v1/chat/completions"
    api_key_avalai = os.getenv("AVALAI_API_KEY")  # Replace with actual key

    url_openai = "https://api.openai.com/v1/chat/completions"
    api_key_openai = os.getenv("OPENAI_API_KEY")  # Replace with actual key

    # Number of requests to make for each API
    num_requests = 60

    # Test prompt
    prompt = "Say hi"

    # Run the tests
    results_avalai = test_api_performance(
        "AvalAI", url_avalai, api_key_avalai, model_name, num_requests, prompt
    )
    results_openai = test_api_performance(
        "OpenAI", url_openai, api_key_openai, model_name, num_requests, prompt
    )

    # Print comparison table
    print_comparison_table(results_avalai, results_openai)

    # Generate visualization
    plot_comparison(results_avalai, results_openai, "api_performance_comparison.png")

    # Save results
    save_results(results_avalai, results_openai, "api_performance_results.json")


if __name__ == "__main__":
    main()

Python

چالش واقعی: اگر نتایج متفاوتی به دست آوردید، به ما بگویید. ما تمام تست‌های خود را عمومی کرده‌ایم؛ شما نیز همین کار را بکنید.

فلسفه ما: چرا این برایمان مهم است؟

بگذارید یک چیز را روشن کنیم: ما yet another api aggregator نیستیم.

کارهایی که ما انجام نمی‌دهیم:

• ❌ افزودن کارمزد به قیمت ارائه‌دهندگان

• ❌ تبلیغات تهاجمی و وعده‌های توخالی

• ❌ ارائه ویژگی‌های بی‌فایده

• ❌ قول‌های بی‌اساس (البته تنظیم دقیق و دستیار را قول داده‌ایم!)

کارهایی که ما انجام می‌دهیم:

• ✅ قیمت‌گذاری ۱۰۰٪ مطابق با ارائه‌دهنده

• ✅ بهینه‌سازی مداوم کارایی

• ✅ شفافیت کامل در آزمون‌های عملکرد

• ✅ تمرکز بر ارائه تجربه‌ای در سطح سازمانی

واقعیت تلخ: بازار پر از تجمیع‌کننده‌های API است که کارمزد دریافت می‌کنند و ارزش افزوده واقعی ارائه نمی‌دهند. ما تصمیم گرفتیم مسیر دیگری را برویم.

حرف آخر: چرا باید باور کنید؟

شما نباید به حرف ما اعتماد کنید. باید به کد اعتماد کنید.

• تمام آزمون‌های عملکرد ما عمومی هستند.

• اسکریپت تست در دسترس همگان است.

• هیچ داده‌ای را دست‌چین نکرده‌ایم.

• نتایج از دو منطقه و دو زمان مختلف به دست آمده‌اند.

اگر تجمیع‌کننده بودن به معنای کندتر بودن بود، ما اکنون این مطلب را نمی‌نوشتیم.

سؤال بحث‌برانگیز: پس چرا همه این کار را انجام نمی‌دهند؟

خب، این سؤال خوبی است. چرا هر واسط API نمی‌تواند این کار را انجام دهد؟

پاسخ ساده: چون کار سختی است. خسته‌کننده است و نیازمند:

• نگهداری از استخرهای اتصالات پیچیده

• نظارت مداوم

• بهینه‌سازی‌های موشکافانه

• صبر و حوصله برای یافتن هر میلی‌ثانیه سربار

پاسخ پیچیده‌تر: چون بیشتر تجمیع‌کننده‌ها بر روی چیزهای دیگری تمرکز کرده‌اند (مانند دریافت هزینه اضافی بیشتر).

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

دعوت به گفتگو

این مطلب برای ایجاد گفتگو نوشته شده است. می‌دانیم که ادعاهای جسورانه‌ای را مطرح می‌کند. از شما می‌خواهیم:

• تست کنید: اسکریپت را اجرا کرده و خودتان نتایج را ببینید.

• سؤال بپرسید: اگر شک دارید، بپرسید.

• به چالش بکشید: اگر نتایج متفاوتی گرفتید، به ما اطلاع دهید.

• بحث کنید: آیا یک تجمیع‌کننده می‌تواند سریع‌تر از دسترسی مستقیم باشد؟

تنها یک درخواست داریم: در هنگام بحث، به داده‌ها نگاه کنید، نه فقط به فرضیات.

لینک‌های مرتبط

• بنچمارک‌های کامل عملکرد

• اسکریپت تست (Python)

• راهنمای بهینه‌سازی تاخیر

• اصول بهترین شیوه‌های اجرا

پی‌نوشت

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

و اگر فکر می‌کنید این تست‌ها دستکاری شده یا گزینشی هستند، کد تست را بردارید، خودتان اجرا کنید و نتایج را به اشتراک بگذارید.

ما اینجا منتظریم. ☕

نویسنده: تیم فنی AvalAI تاریخ: ۱۴۰۴/۰۷/۲۰

پی‌نوشت دوم: اگر تا اینجا خوانده‌اید و هنوز شک دارید، عالی است! شک کردن خوب است. حالا بروید و تست کنید.😉

منبع :
https://avalai.ir/blog/how-is-avalai-api-faster-than-accessing-openai-directly/