gRPC چیست؟ آموزش کامل Protocol Buffers، Streaming و ساخت سرویس gRPC با Python

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

REST API و JSON انتخاب رایجی برای این ارتباط هستند؛ به‌خصوص زمانی که یک API عمومی، وب‌سایت یا اپلیکیشن موبایل می‌سازیم. با این حال، در ارتباط داخلی میان میکروسرویس‌ها ممکن است به قراردادهای دقیق‌تر، پیام‌های کوچک‌تر، تولید خودکار Client و Server و Streaming دوطرفه نیاز داشته باشیم.

gRPC برای پاسخ به چنین نیازهایی طراحی شده است.

در این مقاله یاد می‌گیرید:

• gRPC چیست و چگونه کار می‌کند
• RPC چه تفاوتی با REST دارد
• Protocol Buffers یا Protobuf چیست
• فایل .proto چگونه نوشته می‌شود
• چهار نوع ارتباط در gRPC چه هستند
• gRPC چگونه از HTTP/2 استفاده می‌کند
• چه زمانی gRPC بهتر از REST است
• چگونه با Python یک سرویس gRPC واقعی بسازیم
• چگونه سرویس داخلی gRPC را به API هوش مصنوعی درواره متصل کنیم
• برای استفاده از gRPC در محیط Production چه نکاتی مهم‌اند

gRPC چیست؟

gRPC یک فریم‌ورک متن‌باز و چندزبانه برای Remote Procedure Call یا RPC است. این فریم‌ورک به یک برنامه اجازه می‌دهد متدی را روی یک سرویس دیگر فراخوانی کند؛ به‌گونه‌ای که از دید برنامه‌نویس، فراخوانی آن تا حدی شبیه اجرای یک تابع محلی باشد.

برای مثال، کلاینت می‌تواند متدی با نام زیر را فراخوانی کند:

AnalyzeText(request)

اما این متد در واقع روی سروری دیگر اجرا می‌شود:

Python Client
    ↓
gRPC Request
    ↓
Python، Go، Java یا Node.js Server
    ↓
gRPC Response

در gRPC، ابتدا قرارداد سرویس در یک فایل .proto تعریف می‌شود. سپس ابزارهای gRPC بر اساس همین قرارداد، کدهای Client و Server را برای زبان‌های مختلف تولید می‌کنند.

یک قرارداد ساده:

syntax = "proto3";

service TextAnalyzer {
  rpc AnalyzeText (AnalyzeRequest) returns (AnalyzeResponse);
}

message AnalyzeRequest {
  string text = 1;
}

message AnalyzeResponse {
  string result = 1;
}

این فایل اعلام می‌کند:

• سرویسی به نام TextAnalyzer وجود دارد
• این سرویس متدی به نام AnalyzeText دارد
• ورودی متد از نوع AnalyzeRequest است
• خروجی متد از نوع AnalyzeResponse است
• ورودی شامل یک فیلد متنی است
• خروجی نیز نتیجه تحلیل را برمی‌گرداند

RPC چیست؟

RPC مخفف Remote Procedure Call است. در RPC، کلاینت یک Procedure یا Method را روی سیستم دیگری اجرا می‌کند.

در یک برنامه محلی ممکن است بنویسیم:

result = analyze_text("متن موردنظر")

در RPC نیز تجربه برنامه‌نویسی شبیه همین است:

response = stub.AnalyzeText(
    AnalyzeRequest(text="متن موردنظر")
)

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

• Serialization ورودی
• ارسال پیام روی شبکه
• فراخوانی متد سرور
• دریافت پاسخ
• Deserialize کردن پاسخ
• مدیریت Deadline و Cancellation
• نمایش خطا با Status Code

با وجود ظاهر شبیه تابع محلی، فراخوانی شبکه هیچ‌وقت دقیقاً مانند تابع محلی نیست. شبکه می‌تواند کند یا قطع شود، سرور ممکن است در دسترس نباشد و درخواست ممکن است Timeout شود. بنابراین باید خطاهای شبکه را در طراحی در نظر گرفت.

Protocol Buffers چیست؟

Protocol Buffers یا Protobuf یک روش زبان‌خنثی و مستقل از پلتفرم برای تعریف ساختار داده و Serialize کردن آن است.

Protobuf قالب پیش‌فرض تعریف پیام و سرویس در gRPC است.

در REST APIها معمولاً داده به شکل JSON منتقل می‌شود:

{
  "text": "این متن را خلاصه کن",
  "max_sentences": 3
}

در Protobuf ابتدا Schema داده تعریف می‌شود:

message AnalyzeRequest {
  string text = 1;
  int32 max_sentences = 2;
}

سپس ابزار Protobuf کد لازم برای ساخت، Serialize و Deserialize کردن این پیام را تولید می‌کند.

در Python می‌توان از کلاس تولیدشده استفاده کرد:

request = AnalyzeRequest(
    text="این متن را خلاصه کن",
    max_sentences=3,
)

پیام هنگام ارسال به فرم باینری تبدیل می‌شود. این داده باینری معمولاً نسبت به JSON فشرده‌تر است، اما اندازه و سرعت واقعی باید با Payload و شرایط همان پروژه Benchmark شود.

تفاوت gRPC و Protocol Buffers

gRPC و Protobuf یک مفهوم نیستند.

gRPC به‌صورت پیش‌فرض از Protocol Buffers استفاده می‌کند، اما مفهوم RPC محدود به Protobuf نیست. همچنین می‌توان از Protobuf بدون gRPC برای ذخیره یا انتقال داده استفاده کرد.

gRPC چگونه کار می‌کند؟

چرخه معمول توسعه gRPC به این شکل است:

1. Messageها و Serviceها در فایل .proto تعریف می‌شوند.
2. Compiler بر اساس فایل .proto کد تولید می‌کند.
3. Server رابط تولیدشده را پیاده‌سازی می‌کند.
4. Client از Stub تولیدشده استفاده می‌کند.
5. Client یک Channel با Server می‌سازد.
6. درخواست به پیام Protobuf تبدیل می‌شود.
7. پیام از طریق شبکه برای Server ارسال می‌شود.
8. Server متد مربوط را اجرا می‌کند.
9. پاسخ Serialize و برای Client ارسال می‌شود.
10. Stub پاسخ را به آبجکت زبان برنامه‌نویسی تبدیل می‌کند.

فایل .proto منبع اصلی قرارداد ارتباطی است. بنابراین gRPC معمولاً یک رویکرد Contract-first محسوب می‌شود.

Stub چیست؟

Stub کدی است که بر اساس قرارداد .proto تولید می‌شود و جزئیات ارتباط شبکه را پشت یک رابط برنامه‌نویسی مخفی می‌کند.

نمونه استفاده در Python:

channel = grpc.insecure_channel("localhost:50051")
stub = ai_service_pb2_grpc.AIAnalyzerStub(channel)

response = stub.AnalyzeText(
    ai_service_pb2.AnalyzeRequest(
        text="این متن را تحلیل کن."
    )
)

برنامه‌نویس متد AnalyzeText را فراخوانی می‌کند و Stub موارد زیر را مدیریت می‌کند:

• تبدیل Request به بایت
• ارسال درخواست
• دریافت پاسخ
• تبدیل پاسخ به آبجکت Python
• گزارش خطاهای gRPC

نقش HTTP/2 در gRPC

gRPC معمولاً از HTTP/2 به‌عنوان Transport استفاده می‌کند. HTTP/2 امکانات مهمی در اختیار gRPC قرار می‌دهد:

Multiplexing

چند Request و Response می‌توانند هم‌زمان روی یک Connection منتقل شوند.

Stream

هر RPC می‌تواند از یک HTTP/2 Stream مستقل استفاده کند.

Header Compression

HTTP/2 با فشرده‌سازی Headerها می‌تواند سربار تکراری را کاهش دهد.

Flow Control

Client و Server می‌توانند جریان دریافت داده را کنترل کنند تا گیرنده با حجم بیش از ظرفیت خود مواجه نشود.

ارتباط دوطرفه

HTTP/2 زمینه مناسبی برای Bidirectional Streaming فراهم می‌کند.

این ویژگی‌ها یکی از دلایل مناسب بودن gRPC برای ارتباط سرویس‌به‌سرویس و Streaming هستند.

چهار نوع RPC در gRPC

gRPC چهار الگوی اصلی ارتباط ارائه می‌دهد.

Unary RPC

در Unary RPC، کلاینت یک Request می‌فرستد و یک Response دریافت می‌کند.

تعریف:

rpc AnalyzeText (AnalyzeRequest) returns (AnalyzeResponse);

جریان ارتباط:

یک Request → یک Response

کاربردها:

• دریافت اطلاعات کاربر
• محاسبه قیمت
• تحلیل یک متن
• ایجاد یک سفارش
• دریافت وضعیت یک Job

Unary شبیه Request و Response معمول در REST است.

Server Streaming RPC

در Server Streaming، کلاینت یک Request ارسال می‌کند و Server چند پیام برمی‌گرداند.

تعریف:

rpc StreamSuggestions (SuggestionRequest)
    returns (stream Suggestion);

جریان ارتباط:

یک Request → چند Response

کاربردها:

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

نمونه Client:

responses = stub.StreamSuggestions(request)

for response in responses:
    print(response.text)

Client Streaming RPC

در Client Streaming، کلاینت چند پیام ارسال می‌کند و در پایان یک پاسخ دریافت می‌کند.

تعریف:

rpc AnalyzeBatch (stream TextItem)
    returns (BatchResult);

جریان ارتباط:

چند Request → یک Response

کاربردها:

• آپلود مرحله‌ای داده
• ارسال مجموعه‌ای از Eventها
• جمع‌آوری Metricها
• ارسال بخش‌های یک فایل
• محاسبه نتیجه نهایی روی چند ورودی

Bidirectional Streaming RPC

در Bidirectional Streaming هر دو طرف می‌توانند چند پیام ارسال کنند.

تعریف:

rpc InteractiveReview (stream ReviewMessage)
    returns (stream ReviewMessage);

جریان ارتباط:

چند Request ↔ چند Response

کاربردها:

• چت بلادرنگ
• همکاری زنده
• ارتباط دستگاه‌ها
• پردازش تعاملی
• ارسال و دریافت هم‌زمان Eventها

Stream ورودی و خروجی مستقل هستند. Server مجبور نیست برای هر پیام Client دقیقاً یک پاسخ ارسال کند.

آشنایی با ساختار فایل proto

یک فایل .proto معمولاً با نسخه Syntax آغاز می‌شود:

syntax = "proto3";

سپس Package تعریف می‌شود:

package darvareh.ai.v1;

Package از تداخل نام‌ها جلوگیری می‌کند.

تعریف Message:

message AnalyzeRequest {
  string text = 1;
  int32 max_sentences = 2;
  bool include_category = 3;
}

عددهای 1، 2 و 3 Field Number هستند. این اعداد در فرمت باینری Protobuf برای شناسایی فیلدها استفاده می‌شوند.

انواع داده در Protobuf

برخی نوع‌های پرکاربرد:

نمونه Message تو‌در‌تو:

message Usage {
  int32 input_tokens = 1;
  int32 output_tokens = 2;
}

message AnalyzeResponse {
  string result = 1;
  Usage usage = 2;
}

repeated در Protobuf

برای تعریف لیست از repeated استفاده می‌شود:

message BatchRequest {
  repeated string texts = 1;
}

استفاده در Python:

request = BatchRequest(
    texts=[
        "متن اول",
        "متن دوم",
        "متن سوم",
    ]
)

enum در Protobuf

برای مجموعه‌ای محدود از مقادیر:

enum AnalysisType {
  ANALYSIS_TYPE_UNSPECIFIED = 0;
  ANALYSIS_TYPE_SUMMARY = 1;
  ANALYSIS_TYPE_CLASSIFICATION = 2;
  ANALYSIS_TYPE_KEYWORDS = 3;
}

بهتر است اولین مقدار Enum برابر صفر و نشان‌دهنده حالت نامشخص باشد. این مقدار Default خواهد بود.

optional در Protobuf

اگر لازم است بین «ارسال نشدن فیلد» و «ارسال مقدار پیش‌فرض» تفاوت قائل شوید، می‌توانید از optional استفاده کنید:

message AnalyzeRequest {
  string text = 1;
  optional int32 max_sentences = 2;
}

در این حالت می‌توان Presence فیلد را بررسی کرد.

map در Protobuf

برای ذخیره Key و Value:

message AnalyzeRequest {
  string text = 1;
  map<string, string> metadata = 2;
}

نمونه:

request = AnalyzeRequest(
    text="متن",
    metadata={
        "language": "fa",
        "source": "support",
    },
)

oneof در Protobuf

oneof زمانی استفاده می‌شود که فقط یکی از چند فیلد باید مقدار داشته باشد:

message DocumentInput {
  oneof source {
    string text = 1;
    string file_url = 2;
    bytes file_content = 3;
  }
}

در هر Message فقط یکی از فیلدهای text، file_url یا file_content فعال خواهد بود.

قواعد مهم تغییر Schema در Protobuf

Schema یک قرارداد پایدار میان چند سرویس است. تغییر نادرست آن می‌تواند Clientهای قبلی را خراب کند.

Field Number را تغییر ندهید

نام یک فیلد ممکن است در بعضی شرایط قابل تغییر باشد، اما تغییر Field Number سازگاری باینری را از بین می‌برد.

نامناسب:

string text = 1;

تغییر به:

string text = 5;

Field Number حذف‌شده را دوباره استفاده نکنید

اگر فیلدی حذف شد، شماره آن را Reserve کنید:

message User {
  reserved 2;
  reserved "legacy_name";

  string id = 1;
  string display_name = 3;
}

فیلدهای جدید را با شماره جدید اضافه کنید

نسخه قدیمی Client فیلد ناشناخته را نادیده می‌گیرد و نسخه جدید می‌تواند آن را بخواند.

نوع فیلد را بدون بررسی تغییر ندهید

تغییر string به int32 یا تغییر معنای یک Field ممکن است باعث ناسازگاری شود.

معنی فیلد موجود را تغییر ندهید

حتی اگر نوع فنی یکسان بماند، تغییر معنای Business آن می‌تواند Clientهای قبلی را دچار خطا کند.

تفاوت gRPC و REST API

gRPC و REST هر دو برای ارتباط نرم‌افزارها استفاده می‌شوند، اما فلسفه و تجربه توسعه متفاوتی دارند.

آیا gRPC همیشه سریع‌تر از REST است؟

خیر. نباید بدون Benchmark ادعا کرد gRPC همیشه سریع‌تر است.

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

• Payload باینری
• Connection پایدار
• Multiplexing
• Streaming داخلی
• تولید کد
• قرارداد Type-safe

اما کارایی نهایی به عوامل زیادی بستگی دارد:

• اندازه Payload
• تعداد Requestها
• فاصله شبکه
• نوع داده
• زبان برنامه‌نویسی
• Serialization
• منطق Business
• Database
• تعداد Connectionها
• تنظیمات Load Balancer
• هزینه پردازش سرویس مقصد

اگر بیشتر زمان درخواست صرف Query سنگین Database یا اجرای مدل هوش مصنوعی شود، تفاوت Serialization ممکن است سهم کوچکی از زمان کل باشد.

انتخاب معماری باید بر اساس نیاز واقعی و Benchmark همان سیستم انجام شود.

چه زمانی gRPC انتخاب مناسبی است؟

gRPC معمولاً برای شرایط زیر مناسب است:

• ارتباط داخلی میان میکروسرویس‌ها
• نیاز به Type Safety قوی
• وجود سرویس‌ها با زبان‌های مختلف
• نیاز به تولید خودکار Client و Server
• تعداد زیاد درخواست‌های کوچک
• Server Streaming
• Client Streaming
• Bidirectional Streaming
• ارتباط با Latency پایین
• قرارداد دقیق و قابل بررسی

چه زمانی REST انتخاب بهتری است؟

REST معمولاً در این شرایط انتخاب ساده‌تری است:

• API عمومی برای توسعه‌دهندگان
• اتصال مستقیم Browser
• نیاز به Debug ساده با JSON
• استفاده از Cache استاندارد HTTP
• عملیات Resource-oriented
• تیم کوچک یا پروژه ساده
• سازگاری با ابزارهای فراوان
• نبود نیاز جدی به Streaming دوطرفه

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

مرورگر و اپلیکیشن
        ↓
REST API یا GraphQL
        ↓
Backend
        ↓
gRPC
        ↓
میکروسرویس‌های داخلی

این معماری باعث می‌شود رابط عمومی ساده باقی بماند و ارتباط داخلی از قراردادهای Type-safe gRPC استفاده کند.

ساخت سرویس gRPC با Python و API درواره

در این پروژه یک سرویس داخلی gRPC می‌سازیم که متن را از Client دریافت می‌کند، آن را برای تحلیل به API درواره می‌فرستد و نتیجه را به Client برمی‌گرداند.

معماری پروژه:

Python Client
      ↓ gRPC
AI Analyzer Service
      ↓ HTTPS API
درواره
      ↓
مدل هوش مصنوعی

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

ساختار پروژه

grpc-ai-service/
├── ai_service.proto
├── server.py
├── client.py
├── requirements.txt
└── .env

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

ai_service_pb2.py
ai_service_pb2_grpc.py

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

در Linux و macOS:

python -m venv .venv
source .venv/bin/activate

در Windows PowerShell:

python -m venv .venv
.venv\Scripts\Activate.ps1

نصب وابستگی‌ها

فایل requirements.txt:

grpcio
grpcio-tools
httpx
python-dotenv

نصب:

pip install -r requirements.txt

تعریف قرارداد gRPC

فایل ai_service.proto:

syntax = "proto3";

package darvareh.ai.v1;

service AIAnalyzer {
  rpc AnalyzeText (AnalyzeRequest) returns (AnalyzeResponse);

  rpc StreamSuggestions (SuggestionRequest)
      returns (stream Suggestion);

  rpc AnalyzeBatch (stream BatchItem)
      returns (BatchSummary);

  rpc InteractiveReview (stream ReviewMessage)
      returns (stream ReviewMessage);
}

enum AnalysisType {
  ANALYSIS_TYPE_UNSPECIFIED = 0;
  ANALYSIS_TYPE_SUMMARY = 1;
  ANALYSIS_TYPE_CLASSIFICATION = 2;
  ANALYSIS_TYPE_KEYWORDS = 3;
}

message AnalyzeRequest {
  string text = 1;
  AnalysisType analysis_type = 2;
  optional string instruction = 3;
}

message AnalyzeResponse {
  string result = 1;
  string model = 2;
}

message SuggestionRequest {
  string topic = 1;
  int32 count = 2;
}

message Suggestion {
  int32 index = 1;
  string text = 2;
}

message BatchItem {
  string id = 1;
  string text = 2;
}

message BatchSummary {
  int32 received_count = 1;
  repeated string item_ids = 2;
}

message ReviewMessage {
  string session_id = 1;
  string role = 2;
  string content = 3;
}

در این قرارداد چهار نوع RPC تعریف شده‌اند:

• AnalyzeText: یک Request و یک Response
• StreamSuggestions: یک Request و چند Response
• AnalyzeBatch: چند Request و یک Response
• InteractiveReview: چند Request و چند Response

در پروژه عملی، ابتدا Unary RPC را به درواره متصل می‌کنیم و برای سایر روش‌ها نمونه پیاده‌سازی آموزشی ارائه می‌دهیم.

تولید کد Python از فایل proto

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

python -m grpc_tools.protoc \
  -I. \
  --python_out=. \
  --grpc_python_out=. \
  ai_service.proto

در Windows PowerShell می‌توانید دستور را در یک خط اجرا کنید:

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ai_service.proto

دو فایل تولید می‌شوند:

ai_service_pb2.py
ai_service_pb2_grpc.py

فایل ai_service_pb2.py شامل Messageهای Protobuf است.

فایل ai_service_pb2_grpc.py شامل Stub کلاینت، Servicer سرور و توابع ثبت سرویس است.

این فایل‌ها را دستی ویرایش نکنید. پس از تغییر .proto دوباره آن‌ها را تولید کنید.

تنظیم متغیرهای محیطی

فایل .env:

DARVAREH_API_KEY=YOUR_DARVAREH_API_KEY
DARVAREH_MODEL_ID=YOUR_MODEL_ID

فایل .gitignore:

.env
.venv/
__pycache__/

کلید API را در کد منبع یا Repository عمومی قرار ندهید.

آدرس پایه API درواره:

https://api.darvareh.ir/v1

Endpoint مورد استفاده:

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

پیاده‌سازی gRPC Server

فایل server.py:

import os
from concurrent import futures

import grpc
import httpx
from dotenv import load_dotenv

import ai_service_pb2
import ai_service_pb2_grpc

load_dotenv()

DARVAREH_API_KEY = os.getenv("DARVAREH_API_KEY")
DARVAREH_MODEL_ID = os.getenv("DARVAREH_MODEL_ID")
DARVAREH_URL = "https://api.darvareh.ir/v1/chat/completions"

if not DARVAREH_API_KEY:
    raise RuntimeError("متغیر DARVAREH_API_KEY تنظیم نشده است.")

if not DARVAREH_MODEL_ID:
    raise RuntimeError("متغیر DARVAREH_MODEL_ID تنظیم نشده است.")


ANALYSIS_INSTRUCTIONS = {
    ai_service_pb2.ANALYSIS_TYPE_SUMMARY: (
        "متن را به فارسی روان و دقیق خلاصه کن. "
        "اطلاعات جدیدی به متن اضافه نکن."
    ),
    ai_service_pb2.ANALYSIS_TYPE_CLASSIFICATION: (
        "موضوع اصلی متن را مشخص کن و فقط نام دسته‌بندی "
        "و یک توضیح کوتاه فارسی برگردان."
    ),
    ai_service_pb2.ANALYSIS_TYPE_KEYWORDS: (
        "کلمات کلیدی اصلی متن را به‌صورت یک فهرست کوتاه "
        "و بدون توضیح اضافی برگردان."
    ),
}


class AIAnalyzerService(ai_service_pb2_grpc.AIAnalyzerServicer):
    def AnalyzeText(self, request, context):
        text = request.text.strip()

        if len(text) < 10:
            context.abort(
                grpc.StatusCode.INVALID_ARGUMENT,
                "متن باید حداقل ۱۰ کاراکتر داشته باشد.",
            )

        if len(text) > 20_000:
            context.abort(
                grpc.StatusCode.INVALID_ARGUMENT,
                "طول متن از محدودیت این سرویس بیشتر است.",
            )

        base_instruction = ANALYSIS_INSTRUCTIONS.get(
            request.analysis_type
        )

        if not base_instruction:
            context.abort(
                grpc.StatusCode.INVALID_ARGUMENT,
                "نوع تحلیل معتبر نیست.",
            )

        instruction = base_instruction

        if request.HasField("instruction"):
            custom_instruction = request.instruction.strip()

            if custom_instruction:
                instruction += (
                    "\nدر صورت سازگار بودن با هدف اصلی، "
                    f"این توضیح را نیز رعایت کن: {custom_instruction}"
                )

        request_body = {
            "model": DARVAREH_MODEL_ID,
            "messages": [
                {
                    "role": "system",
                    "content": instruction,
                },
                {
                    "role": "user",
                    "content": text,
                },
            ],
        }

        headers = {
            "Authorization": f"Bearer {DARVAREH_API_KEY}",
            "Content-Type": "application/json",
        }

        timeout = httpx.Timeout(
            connect=10.0,
            read=60.0,
            write=20.0,
            pool=10.0,
        )

        try:
            with httpx.Client(timeout=timeout) as client:
                response = client.post(
                    DARVAREH_URL,
                    headers=headers,
                    json=request_body,
                )

            if response.status_code == 429:
                context.abort(
                    grpc.StatusCode.RESOURCE_EXHAUSTED,
                    "محدودیت تعداد درخواست سرویس بالادستی فعال شده است.",
                )

            if response.status_code in {502, 503, 504}:
                context.abort(
                    grpc.StatusCode.UNAVAILABLE,
                    "سرویس هوش مصنوعی موقتاً در دسترس نیست.",
                )

            if response.status_code >= 400:
                context.abort(
                    grpc.StatusCode.FAILED_PRECONDITION,
                    "درخواست توسط سرویس بالادستی پذیرفته نشد.",
                )

            body = response.json()
            result = body["choices"][0]["message"]["content"].strip()

            if not result:
                context.abort(
                    grpc.StatusCode.INTERNAL,
                    "پاسخ سرویس هوش مصنوعی خالی بود.",
                )

        except httpx.TimeoutException:
            context.abort(
                grpc.StatusCode.DEADLINE_EXCEEDED,
                "سرویس هوش مصنوعی در زمان تعیین‌شده پاسخ نداد.",
            )
        except httpx.RequestError:
            context.abort(
                grpc.StatusCode.UNAVAILABLE,
                "ارتباط با سرویس هوش مصنوعی برقرار نشد.",
            )
        except (ValueError, KeyError, IndexError, TypeError):
            context.abort(
                grpc.StatusCode.INTERNAL,
                "ساختار پاسخ سرویس بالادستی معتبر نبود.",
            )

        return ai_service_pb2.AnalyzeResponse(
            result=result,
            model=DARVAREH_MODEL_ID,
        )

    def StreamSuggestions(self, request, context):
        count = request.count

        if count < 1 or count > 20:
            context.abort(
                grpc.StatusCode.INVALID_ARGUMENT,
                "تعداد پیشنهاد باید بین ۱ تا ۲۰ باشد.",
            )

        topic = request.topic.strip()

        if not topic:
            context.abort(
                grpc.StatusCode.INVALID_ARGUMENT,
                "موضوع نمی‌تواند خالی باشد.",
            )

        for index in range(1, count + 1):
            if not context.is_active():
                return

            yield ai_service_pb2.Suggestion(
                index=index,
                text=f"پیشنهاد شماره {index} برای موضوع {topic}",
            )

    def AnalyzeBatch(self, request_iterator, context):
        item_ids = []

        for item in request_iterator:
            if not context.is_active():
                break

            if item.id and item.text.strip():
                item_ids.append(item.id)

            if len(item_ids) > 1000:
                context.abort(
                    grpc.StatusCode.RESOURCE_EXHAUSTED,
                    "تعداد آیتم‌های Batch از محدودیت بیشتر است.",
                )

        return ai_service_pb2.BatchSummary(
            received_count=len(item_ids),
            item_ids=item_ids,
        )

    def InteractiveReview(self, request_iterator, context):
        for message in request_iterator:
            if not context.is_active():
                return

            content = message.content.strip()

            if not content:
                continue

            yield ai_service_pb2.ReviewMessage(
                session_id=message.session_id,
                role="assistant",
                content=f"پیام دریافت شد: {content}",
            )


def serve():
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        options=[
            (
                "grpc.max_receive_message_length",
                4 * 1024 * 1024,
            ),
            (
                "grpc.max_send_message_length",
                4 * 1024 * 1024,
            ),
        ],
    )

    ai_service_pb2_grpc.add_AIAnalyzerServicer_to_server(
        AIAnalyzerService(),
        server,
    )

    server.add_insecure_port("[::]:50051")
    server.start()

    print("gRPC server is running on port 50051")

    try:
        server.wait_for_termination()
    except KeyboardInterrupt:
        server.stop(grace=5)


if __name__ == "__main__":
    serve()

این Server چهار الگوی ارتباطی را پیاده‌سازی می‌کند. فقط متد AnalyzeText به API درواره متصل است و سه متد دیگر برای نمایش نحوه کار Streaming نوشته شده‌اند.

در یک پروژه واقعی می‌توانید متدهای Streaming را نیز به Worker، Database یا سرویس پردازشی مناسب متصل کنید.

اجرای gRPC Server

python server.py

خروجی:

gRPC server is running on port 50051

ساخت gRPC Client

فایل client.py:

import grpc

import ai_service_pb2
import ai_service_pb2_grpc


def test_unary(stub):
    request = ai_service_pb2.AnalyzeRequest(
        text=(
            "gRPC یک فریم‌ورک RPC چندزبانه است که برای "
            "ارتباط میان سرویس‌ها استفاده می‌شود. این فناوری "
            "به‌صورت پیش‌فرض از Protocol Buffers برای تعریف "
            "قرارداد و انتقال داده استفاده می‌کند."
        ),
        analysis_type=ai_service_pb2.ANALYSIS_TYPE_SUMMARY,
        instruction="پاسخ بیشتر از دو جمله نباشد.",
    )

    response = stub.AnalyzeText(
        request,
        timeout=70,
    )

    print("Result:")
    print(response.result)
    print("Model:", response.model)


def test_server_streaming(stub):
    request = ai_service_pb2.SuggestionRequest(
        topic="بهبود مستندات API",
        count=3,
    )

    print("\nServer streaming:")

    for suggestion in stub.StreamSuggestions(
        request,
        timeout=10,
    ):
        print(
            suggestion.index,
            suggestion.text,
        )


def batch_items():
    items = [
        ("item_1", "متن اول"),
        ("item_2", "متن دوم"),
        ("item_3", "متن سوم"),
    ]

    for item_id, text in items:
        yield ai_service_pb2.BatchItem(
            id=item_id,
            text=text,
        )


def test_client_streaming(stub):
    response = stub.AnalyzeBatch(
        batch_items(),
        timeout=10,
    )

    print("\nClient streaming:")
    print("Received:", response.received_count)
    print("IDs:", list(response.item_ids))


def review_messages():
    messages = [
        "مقدمه مقاله را کوتاه‌تر کن.",
        "یک مثال عملی اضافه کن.",
        "جمع‌بندی را واضح‌تر بنویس.",
    ]

    for content in messages:
        yield ai_service_pb2.ReviewMessage(
            session_id="session_101",
            role="user",
            content=content,
        )


def test_bidirectional_streaming(stub):
    print("\nBidirectional streaming:")

    responses = stub.InteractiveReview(
        review_messages(),
        timeout=10,
    )

    for response in responses:
        print(
            response.role,
            response.content,
        )


def main():
    try:
        with grpc.insecure_channel(
            "localhost:50051"
        ) as channel:
            stub = ai_service_pb2_grpc.AIAnalyzerStub(
                channel
            )

            test_unary(stub)
            test_server_streaming(stub)
            test_client_streaming(stub)
            test_bidirectional_streaming(stub)

    except grpc.RpcError as error:
        print("gRPC request failed")
        print("Status:", error.code())
        print("Details:", error.details())


if __name__ == "__main__":
    main()

Client را اجرا کنید:

python client.py

در بخش Unary، متن به Server ارسال می‌شود. Server درخواست را به API درواره می‌فرستد و نتیجه مدل را از طریق gRPC به Client برمی‌گرداند.

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

چرا Deadline ضروری است؟

فراخوانی شبکه نباید برای همیشه منتظر بماند.

در Client مقدار Deadline تعیین کردیم:

response = stub.AnalyzeText(
    request,
    timeout=70,
)

اگر پاسخ در زمان تعیین‌شده نرسد، Client خطای زیر دریافت می‌کند:

DEADLINE_EXCEEDED

Server نیز می‌تواند فعال بودن Context را بررسی کند:

if not context.is_active():
    return

اگر Client درخواست را لغو کرده باشد یا Deadline تمام شده باشد، ادامه دادن پردازش ممکن است منابع را بیهوده مصرف کند.

Deadline بهتر است در طول زنجیره سرویس‌ها منتقل شود:

Client Deadline: 10s
Service A: حداکثر 9s
Service B: حداکثر 8s
Database: حداکثر 2s

هر سرویس باید بخشی از زمان باقی‌مانده را برای پردازش و پاسخ‌گویی حفظ کند.

Status Codeهای gRPC

gRPC مجموعه‌ای از Status Codeهای مستقل از کدهای وضعیت HTTP ارائه می‌کند.

برای خطاهای قابل پیش‌بینی از Status مناسب استفاده کنید و تمام خطاها را به INTERNAL تبدیل نکنید.

Metadata در gRPC

Metadata مشابه Header در HTTP است و می‌تواند اطلاعات جانبی Request را منتقل کند.

نمونه Client:

metadata = [
    ("authorization", "Bearer INTERNAL_TOKEN"),
    ("x-request-id", "req_8127"),
]

response = stub.AnalyzeText(
    request,
    metadata=metadata,
    timeout=70,
)

نمونه دریافت در Server:

metadata = dict(
    context.invocation_metadata()
)

request_id = metadata.get("x-request-id")

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

• اطلاعات هویتی
• Request ID
• Trace ID
• نسخه Client
• اطلاعات Locale
• Tenant ID

داده اصلی Business را بهتر است داخل Message قرار دهید، نه Metadata.

Channel را برای هر Request نسازید

ایجاد Channel جدید برای هر درخواست باعث از دست رفتن مزایای Connection پایدار می‌شود.

طراحی نامناسب:

def call_service():
    channel = grpc.insecure_channel("localhost:50051")
    stub = AIAnalyzerStub(channel)
    return stub.AnalyzeText(request)

بهتر است Channel و Stub را Reuse کنید:

channel = grpc.insecure_channel("localhost:50051")
stub = AIAnalyzerStub(channel)

def call_service(request):
    return stub.AnalyzeText(
        request,
        timeout=10,
    )

در برنامه‌های واقعی، Lifecycle اتصال باید با Lifecycle برنامه هماهنگ شود.

gRPC در Browser

مرورگرها معمولاً نمی‌توانند مستقیماً مانند Backendهای Native از تمام قابلیت‌های استاندارد gRPC استفاده کنند.

برای Browser می‌توان از gRPC-Web و یک Proxy سازگار استفاده کرد. با این حال، تمام قابلیت‌های Streaming در همه معماری‌های مرورگری به یک شکل در دسترس نیستند.

برای بسیاری از محصولات، معماری زیر ساده‌تر است:

Browser
   ↓ REST، GraphQL یا gRPC-Web
Backend for Frontend
   ↓ gRPC
Internal Services

اگر API برای کاربران عمومی و توسعه‌دهندگان بیرونی ارائه می‌شود، REST و JSON اغلب تجربه ساده‌تری ایجاد می‌کنند.

Health Check در gRPC

فقط باز بودن Port به معنی سالم بودن سرویس نیست. Health Check باید مشخص کند سرویس آماده دریافت درخواست است یا خیر.

می‌توانید از پروتکل استاندارد gRPC Health Checking استفاده کنید تا Load Balancer یا سیستم Orchestration وضعیت سرویس را بررسی کند.

وضعیت‌های متداول:

UNKNOWN
SERVING
NOT_SERVING
SERVICE_UNKNOWN

Health Check نباید برای هر درخواست Query سنگین Database یا فراخوانی مدل هوش مصنوعی انجام دهد. هدف آن بررسی آمادگی عملیاتی سرویس است.

Interceptor چیست؟

Interceptor در gRPC مشابه Middleware در فریم‌ورک‌های وب است.

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

• Logging
• Metrics
• Request ID
• بررسی Metadata
• محدودیت درخواست
• تبدیل خطا
• Tracing

منطق مشترک را به‌جای تکرار در تمام Methodها می‌توان در Interceptor قرار داد.

Retry در gRPC

Retry باید با احتیاط انجام شود. اگر متدی اثر جانبی دارد، تکرار آن ممکن است عملیات را چند بار اجرا کند.

Retry معمولاً برای خطاهای موقت مانند UNAVAILABLE مناسب‌تر است، اما باید موارد زیر را در نظر گرفت:

• Idempotent بودن Method
• تعداد محدود تلاش
• Exponential Backoff
• Jitter
• Deadline کلی
• ظرفیت Server
• جلوگیری از Retry Storm

برای عملیاتی مانند ایجاد سفارش یا کسر اعتبار، باید شناسه Idempotency یا سازوکار Deduplication وجود داشته باشد.

gRPC و Load Balancing

در معماری ساده، Load Balancer می‌تواند اتصال gRPC را میان Serverها توزیع کند. اما چون HTTP/2 Connection ممکن است طولانی‌مدت باقی بماند، طراحی Load Balancing باید با رفتار gRPC سازگار باشد.

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

• Proxy یا Load Balancer آگاه از HTTP/2
• Client-side Load Balancing
• Service Discovery
• DNS-based Discovery
• Service Mesh

صرف قرار دادن یک Load Balancer قدیمی جلوی gRPC الزاماً توزیع متعادل Requestها را تضمین نمی‌کند.

Observability در gRPC

برای هر RPC بهتر است اطلاعات زیر ثبت شود:

• نام Service
• نام Method
• Status Code
• مدت اجرا
• Request ID
• Trace ID
• اندازه Request
• اندازه Response
• Deadline
• تعداد Retry
• نام سرویس بالادستی
• زمان پاسخ سرویس بالادستی

از ثبت API Key، محتوای محرمانه یا متن کامل کاربران در Log خودداری کنید؛ مگر با سیاست روشن، نیاز واقعی و کنترل دسترسی مناسب.

شاخص‌های مهم:

RPC Request Rate
RPC Error Rate
RPC Latency
Active Streams
Message Size
Deadline Exceeded Rate
Unavailable Rate
Upstream Latency

تست سرویس gRPC

تست‌ها را می‌توان در چند سطح انجام داد.

تست Unit

متد Service را بدون اجرای Server کامل آزمایش کنید.

تست Integration

Server را اجرا و از طریق Stub واقعی فراخوانی کنید.

تست Contract

بررسی کنید Client و Server از نسخه‌های سازگار فایل .proto استفاده می‌کنند.

تست Streaming

شرایط زیر را آزمایش کنید:

• Stream خالی
• Stream طولانی
• قطع ارتباط
• Cancellation
• Deadline
• پیام نامعتبر
• Client کند
• Server کند

تست Failure

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

• در دسترس نبودن سرویس بالادستی
• Timeout
• پاسخ نامعتبر
• Resource Exhaustion
• Restart شدن Server
• Retry هم‌زمان چند Client

اشتباهات رایج در gRPC

تغییر دادن Field Number

Field Number بخشی از قرارداد باینری است و نباید بدون Migration تغییر کند.

استفاده مجدد از شماره فیلد حذف‌شده

شماره حذف‌شده را با reserved محافظت کنید.

ساخت Channel برای هر درخواست

Channel باید تا حد امکان Reuse شود.

نداشتن Deadline

درخواست بدون Deadline می‌تواند مدت زیادی منابع Client و Server را اشغال کند.

فرض کردن فراخوانی Remote مانند تابع Local

شبکه، Timeout، Partial Failure و Retry باید در طراحی دیده شوند.

ارسال Messageهای بسیار بزرگ

gRPC برای انتقال بی‌محدودیت فایل‌های بسیار بزرگ طراحی نشده است. برای فایل‌های بزرگ می‌توان از Object Storage و ارسال Reference استفاده کرد یا Chunking کنترل‌شده ساخت.

استفاده اجباری از gRPC برای API عمومی

gRPC برای همه پروژه‌ها بهترین انتخاب نیست. REST ممکن است برای مصرف‌کنندگان عمومی ساده‌تر باشد.

نادیده گرفتن Browser

اگر Client اصلی مرورگر است، محدودیت‌های gRPC-Web و Proxy را پیش از انتخاب معماری بررسی کنید.

Retry عملیات غیر Idempotent

این کار ممکن است عملیات را چند بار اجرا کند.

قرار دادن API Key در Client عمومی

اگر سرویس gRPC توسط اپلیکیشن قابل توزیع استفاده شود، قرار دادن کلید در برنامه Client می‌تواند آن را در معرض استخراج قرار دهد. کلید درواره باید در Backend نگهداری شود.

برگرداندن خروجی مدل بدون اعتبارسنجی

اگر نتیجه هوش مصنوعی وارد Workflow، Database یا سیستم دیگری می‌شود، ساختار و مقادیر آن را بررسی کنید.

چک‌لیست Production برای gRPC

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

• قرارداد .proto مشخص و نسخه‌بندی شده است
• Package مناسب تعریف شده است
• Field Numberها پایدار هستند
• فیلدهای حذف‌شده Reserve شده‌اند
• Messageها بیش از حد بزرگ نیستند
• Deadline روی تمام فراخوانی‌ها وجود دارد
• Cancellation در پردازش‌های طولانی رعایت می‌شود
• Channel و Stub دوباره استفاده می‌شوند
• Status Codeها معنای درست دارند
• Retry فقط برای خطاهای مناسب انجام می‌شود
• عملیات حساس Idempotent یا Deduplicate شده‌اند
• Health Check پیاده‌سازی شده است
• Shutdown به‌صورت Graceful انجام می‌شود
• Load Balancer از HTTP/2 پشتیبانی می‌کند
• Logging و Metrics فعال هستند
• Request ID و Trace ID منتقل می‌شوند
• اطلاعات محرمانه وارد Log نمی‌شوند
• محدودیت اندازه پیام تعیین شده است
• ظرفیت Thread Pool یا Async Runtime بررسی شده است
• تست سازگاری Schema وجود دارد
• سناریوهای Streaming و قطع ارتباط آزمایش شده‌اند
• کلید API فقط در Backend نگهداری می‌شود
• خروجی مدل هوش مصنوعی اعتبارسنجی می‌شود

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

gRPC به زبان ساده چیست؟

gRPC یک روش برای فراخوانی متدهای یک سرویس از طریق شبکه است. قرارداد متدها و پیام‌ها معمولاً در فایل .proto تعریف و کد Client و Server به‌صورت خودکار تولید می‌شود.

gRPC مخفف چیست؟

بر اساس FAQ رسمی پروژه، gRPC به‌صورت بازگشتی به معنی gRPC Remote Procedure Calls است.

آیا gRPC یک پروتکل است؟

gRPC یک فریم‌ورک RPC با قراردادها و قواعد مشخص است که معمولاً از HTTP/2 برای انتقال و Protocol Buffers برای تعریف و Serialize کردن پیام‌ها استفاده می‌کند.

آیا gRPC جایگزین REST است؟

نه در تمام پروژه‌ها. gRPC برای ارتباط داخلی، Streaming و قراردادهای Type-safe بسیار مناسب است. REST برای API عمومی، Browser و ارتباط JSON ساده‌تر است. بسیاری از سیستم‌ها از هر دو استفاده می‌کنند.

آیا gRPC همیشه از Protobuf استفاده می‌کند؟

Protobuf روش پیش‌فرض و رایج gRPC است، اما مفهوم gRPC از نظر معماری الزاماً به یک Serialization خاص محدود نیست. بیشتر ابزارها و آموزش‌های gRPC بر Protobuf متمرکزند.

آیا Protobuf از JSON سریع‌تر است؟

Protobuf در بسیاری از Payloadها کوچک‌تر و سریع‌تر Serialize می‌شود، اما نتیجه به ساختار داده، زبان و شرایط سیستم بستگی دارد. برای تصمیم معماری باید Benchmark واقعی انجام شود.

آیا gRPC برای میکروسرویس مناسب است؟

بله. Contract-first بودن، تولید Stub، پشتیبانی چندزبانه و Streaming باعث شده‌اند gRPC یکی از گزینه‌های مهم برای ارتباط میان میکروسرویس‌ها باشد.

آیا می‌توان gRPC را از Browser فراخوانی کرد؟

برای این کار معمولاً از gRPC-Web و یک Proxy سازگار استفاده می‌شود. اگر مخاطب اصلی Browser است، REST یا GraphQL ممکن است ساده‌تر باشد.

تفاوت Server Streaming و Bidirectional Streaming چیست؟

در Server Streaming، Client یک پیام می‌فرستد و Server چند پیام برمی‌گرداند. در Bidirectional Streaming هر دو طرف می‌توانند چند پیام ارسال کنند.

آیا gRPC برای API هوش مصنوعی مناسب است؟

برای ارتباط داخلی سرویس‌های هوش مصنوعی، Streaming و فراخوانی‌های Type-safe می‌تواند مناسب باشد. API عمومی مدل‌ها معمولاً REST یا HTTP Streaming ارائه می‌شود. می‌توان یک Gateway یا سرویس داخلی gRPC را به API هوش مصنوعی متصل کرد.

آیا API درواره gRPC است؟

اتصال عمومی درواره از طریق API مبتنی بر HTTPS و ساختار سازگار با APIهای رایج هوش مصنوعی انجام می‌شود. در معماری این مقاله، سرویس داخلی شما gRPC است و همان سرویس برای دسترسی به مدل، API درواره را فراخوانی می‌کند.

Base URL درواره چیست؟

https://api.darvareh.ir/v1

Endpoint مربوط به Chat Completions:

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

Model ID و قیمت مدل‌ها را از کجا ببینیم؟

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

جمع‌بندی

gRPC یک فریم‌ورک مدرن برای ارتباط Remote Procedure Call است که امکان تعریف قراردادهای دقیق، تولید خودکار Client و Server و اجرای Unary و Streaming RPC را فراهم می‌کند.

در gRPC، فایل .proto نقش قرارداد اصلی را دارد. Protocol Buffers ساختار Messageها را تعریف می‌کند، کدهای لازم برای زبان‌های مختلف تولید می‌شوند و ارتباط معمولاً روی HTTP/2 انجام می‌شود.

gRPC به‌خصوص برای ارتباط داخلی میکروسرویس‌ها، سیستم‌های چندزبانه، Streaming و Requestهای پرتعداد مناسب است. با این حال، برای API عمومی، Browser و سناریوهایی که سادگی JSON اهمیت بیشتری دارد، REST می‌تواند انتخاب بهتری باشد.

در پروژه عملی این مقاله، یک سرویس gRPC با Python ساختیم که متن را از Client دریافت و برای پردازش به API درواره ارسال می‌کند. این معماری اجازه می‌دهد سرویس‌های داخلی از قرارداد Type-safe gRPC استفاده کنند و در عین حال از طریق یک API واحد به مدل‌های مختلف هوش مصنوعی دسترسی داشته باشند.

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

منابع

• وب‌سایت رسمی gRPC
• معرفی رسمی gRPC
• مفاهیم اصلی، معماری و چرخه حیات gRPC
• راهنمای رسمی زبان Proto3
• آموزش رسمی gRPC برای Python
• راهنمای Performance در gRPC
• استاندارد HTTP/2، RFC 9113

مقالات مرتبط

• ساخت API هوش مصنوعی آماده محیط Production
• راهنمای جامع API Gateway
• AI Gateway چیست و چه تفاوتی با API Gateway دارد؟
• API سازگار با OpenAI چیست؟
• معماری Multi-model و Multi-provider برای سرویس‌های هوش مصنوعی
• پردازش غیرهم‌زمان API هوش مصنوعی با Celery، Redis و Worker
• هوش مصنوعی با Python و API؛ راهنمای کامل

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