Ollama: تكامل Python SDK
Python SDK هو المفتاح الذي يفتح باب الذكاء الاصطناعي المحلي — ثلاثة أسطر كود، انتقال سلس من السكربت إلى التطبيق.
💡 نصيحة: مكتبة Python SDK (
ollama) هي في الأساس غلاف حول واجهة Ollama REST API — طريقة chat() تُغلف /api/chat، وgenerate() تُغلف /api/generate، وlist() تُغلف /api/tags. فهم هذه العلاقة يساعد في استكشاف الأخطاء: عندما يُطلق SDK خطأً، يمكنك استخدام curl لاستدعاء API مباشرة لتحديد ما إذا كانت المشكلة من SDK أو من خدمة Ollama.
📋 المتطلبات المسبقة: يجب أن تتقن ما يلي أولًا
- الدرس 5: أساسيات REST API
1. ماذا ستتعلم
- تثبيت مكتبة ollama في Python وواجهات API المتزامنة/غير المتزامنة
- استخدام
chat()/generate()/list() - تنفيذ الاستجابات المتدفقة غير المتزامنة للإخراج الفوري
- توضيحات أنواع Python ومعالجة الأخطاء
- غلاف Python لـ SupportBot V1 لأليس
2. قصة حقيقية من مؤسس SaaS
⚠️ تحذير: عندما يستدعي Python SDK واجهة Ollama API مباشرة، لا تحتوي Ollama على مصادقة مدمجة. إذا كانت Ollama مرتبطة بـ
0.0.0.0، فقد يُستغل معلومات اتصال SDK (مثلاً، http://your-server:11434) من قبل آخرين. في الإنتاج، أضف دائمًا طبقة مصادقة.
(1) المشكلة: سكربتات Shell غير كافية
بنت أليس نموذجًا أوليًا لـ SupportBot باستخدام curl، لكن سكربتات Shell صعبة الصيانة لحالة المحادثة ومعالجة الأخطاء والتكامل مع خدمات الويب. تحتاج لغة برمجة مناسبة لبناء تطبيقات بمستوى الإنتاج.
(2) الحل: تكامل بثلاثة أسطر عبر Python SDK
PYTHON
import ollama
response = ollama.chat(model='qwen2.5', messages=[
{'role': 'user', 'content': 'Hello'}
])
print(response['message']['content'])
3. التثبيت ونظرة عامة على API
ℹ️ معلومة: يتصل Python SDK افتراضيًا بـ
http://localhost:11434، بدون حاجة لإعداد إضافي. إذا كانت Ollama تعمل على عنوان مختلف، يمكنك تجاوزه بضبط المتغير البيئي OLLAMA_HOST أو تحديد Client(host='http://...') في الكود.
💡 نصيحة: يُوصى بطريقة
chat() في Python SDK على generate() — chat يدعم الحوار متعدد الأدوار (مصفوفة messages)، بينما generate يدعم الطلقة الواحدة فقط. حتى للأسئلة أحادية الطلقة، تمييز الأدوار في chat (system/user/assistant) ينتج جودة إخراج أفضل.
(1) التثبيت والتحقق من الاتصال
BASH
# تثبيت حزمة ollama في Python
pip install ollama
# التحقق من الاتصال بخادم Ollama
python3 -c "import ollama; print(ollama.list())"
(2) مقارنة واجهة API المتزامنة وغير المتزامنة
⚠️ ملاحظة: عند استخدام واجهة API غير المتزامنة (
AsyncClient)، يجب أن تسبق جميع الاستدعاءات بـ await، مثلاً، await client.chat(...). نسيان await يُرجع كائن coroutine بدلاً من النتيجة الفعلية — البرنامج لن يُخطئ لكنه لن يُنتج إخراجًا صحيحًا. في أُطر العمل غير المتزامنة مثل FastAPI، يجب استخدام واجهة API غير المتزامنة؛ وإلا فإنها تحجب حلقة الأحداث وتؤثر على الأداء المتزامن.
| البُعد | واجهة API المتزامنة | واجهة API غير المتزامنة |
|---|---|---|
| الوحدة | ollama |
ollama (AsyncClient) |
| نمط الاستدعاء | ollama.chat() |
await client.chat() |
| الحجب | يحجب الخيط الحالي | غير حاجب، متزامن |
| حالة الاستخدام | سكربتات، أدوات بسيطة | خدمات ويب، معالجة متزامنة |
| دعم المتدفقة | for chunk in stream |
async for chunk in stream |
▶ مثال 1: استدعاءات أساسية متزامنة وغير متزامنة
PYTHON
import ollama
import asyncio
# استدعاء متزامن
def sync_chat():
response = ollama.chat(
model='qwen2.5',
messages=[{'role': 'user', 'content': 'Hello!'}]
)
print(response['message']['content'])
# استدعاء غير متزامن
async def async_chat():
client = ollama.AsyncClient()
response = await client.chat(
model='qwen2.5',
messages=[{'role': 'user', 'content': 'Hello!'}]
)
print(response['message']['content'])
sync_chat()
asyncio.run(async_chat())
الإخراج:
TEXT
# Function defined successfully
4. تفصيل الطرق الأساسية
(1) طريقة chat()
| المعلمة | النوع | الوصف |
|---|---|---|
model |
str | اسم النموذج |
messages |
list[dict] | قائمة الرسائل، كل منها يحتوي role/content |
stream |
bool | هل الإخراج متدفق |
format |
str | تنسيق الإخراج: json |
options |
dict | معلمات الاستنتاج (temperature، إلخ) |
keep_alive |
str | مدة بقاء النموذج في الذاكرة |
(2) طريقة generate()
| المعلمة | النوع | الوصف |
|---|---|---|
model |
str | اسم النموذج |
prompt |
str | نص الموجه |
system |
str | موجه النظام |
stream |
bool | هل الإخراج متدفق |
options |
dict | معلمات الاستنتاج |
▶ مثال 2: مقارنة chat مقابل generate
PYTHON
import ollama
# chat(): حوار متعدد الأدوار بسجل رسائل
response = ollama.chat(
model='qwen2.5',
messages=[
{'role': 'system', 'content': 'You are a SQL expert.'},
{'role': 'user', 'content': 'Write a query for top 5 customers'}
],
stream=False,
options={'temperature': 0.3}
)
print('chat:', response['message']['content'])
# generate(): إنشاء نص أحادي الطلقة
response = ollama.generate(
model='qwen2.5',
prompt='Write a haiku about debugging',
system='You are a poet.',
stream=False
)
print('generate:', response['response'])
الإخراج:
TEXT
chat:
generate:
5. تنفيذ الاستجابات المتدفقة
(1) مبدأ الإخراج المتدفق
sequenceDiagram
participant P as Python App
participant O as Ollama Server
P->>O: chat(stream=True)
loop Each token chunk
O-->>P: chunk {"content": "word"}
P->>P: print(word, end="")
end
O-->>P: chunk {"done": true}
▶ مثال 3: إخراج متدفق متزامن
PYTHON
import ollama
# إخراج محادثة متدفقة فوريًا
stream = ollama.chat(
model='qwen2.5',
messages=[{'role': 'user', 'content': 'Explain RAG in 3 sentences'}],
stream=True
)
for chunk in stream:
content = chunk['message']['content']
print(content, end='', flush=True)
print() # سطر جديد في النهاية
الإخراج:
TEXT
# Execution successful
▶ مثال 4: إخراج متدفق غير متزامن
PYTHON
import ollama
import asyncio
async def stream_chat():
client = ollama.AsyncClient()
stream = await client.chat(
model='qwen2.5',
messages=[{'role': 'user', 'content': 'Tell me about Ollama'}],
stream=True
)
async for chunk in stream:
content = chunk['message']['content']
print(content, end='', flush=True)
print()
asyncio.run(stream_chat())
الإخراج:
TEXT
# Function defined successfully
6. معالجة الأخطاء وتوضيحات الأنواع
(1) أنواع الأخطاء الشائعة
| الخطأ | شرط التفعيل | المعالجة |
|---|---|---|
ConnectionError |
خدمة Ollama لا تعمل | ابدأ الخدمة أو أعد المحاولة |
ResponseError |
نموذج غير موجود / معلمات غير صالحة | تحقق من اسم النموذج والمعلمات |
TimeoutError |
انتهت مهلة الاستنتاج | قلل num_ctx أو زد المهلة |
JSONDecodeError |
شذوذ في إخراج format=json | أضف تحقق JSON وأعد المحاولة |
▶ مثال 5: معالجة أخطاء متينة
PYTHON
import ollama
import json
from typing import Optional
def safe_chat(
model: str,
messages: list[dict],
temperature: float = 0.3,
max_retries: int = 3
) -> Optional[str]:
"""Chat with error handling and retries."""
for attempt in range(max_retries):
try:
response = ollama.chat(
model=model,
messages=messages,
stream=False,
options={'temperature': temperature}
)
return response['message']['content']
except ConnectionError:
print(f"Connection failed (attempt {attempt + 1})")
if attempt == max_retries - 1:
return None
except ollama.ResponseError as e:
print(f"API error: {e.error}")
return None
except Exception as e:
print(f"Unexpected error: {e}")
if attempt == max_retries - 1:
return None
return None
# الاستخدام
result = safe_chat('qwen2.5', [
{'role': 'user', 'content': 'What is your return policy?'}
])
if result:
print(result)
else:
print("Failed to get response")
الإخراج:
TEXT
Failed to get response
7. مثال شامل: غلاف Python لـ SupportBot V1
PYTHON
# ============================================
# شامل: SupportBot V1
# غلاف Python لخدمة عملاء التجارة الإلكترونية
# ============================================
import ollama
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class SupportBot:
model: str = "qwen2.5"
temperature: float = 0.4
max_history: int = 10
system_prompt: str = (
"You are SupportBot, an e-commerce customer service agent. "
"Be polite, concise, and helpful. "
"If unsure, say 'Let me connect you with a human agent.'"
)
messages: list[dict] = field(default_factory=list)
def __post_init__(self):
self.messages = [
{"role": "system", "content": self.system_prompt}
]
def chat(self, user_input: str) -> str:
self.messages.append({"role": "user", "content": user_input})
try:
response = ollama.chat(
model=self.model,
messages=self.messages[-self.max_history:],
stream=False,
options={"temperature": self.temperature}
)
assistant_msg = response["message"]["content"]
self.messages.append({"role": "assistant", "content": assistant_msg})
return assistant_msg
except Exception as e:
self.messages.pop() # Remove failed user message
return f"Error: {str(e)}"
def stream_chat(self, user_input: str):
self.messages.append({"role": "user", "content": user_input})
full_response = []
try:
stream = ollama.chat(
model=self.model,
messages=self.messages[-self.max_history:],
stream=True,
options={"temperature": self.temperature}
)
for chunk in stream:
content = chunk["message"]["content"]
full_response.append(content)
print(content, end="", flush=True)
print()
self.messages.append({"role": "assistant", "content": "".join(full_response)})
except Exception as e:
print(f"\nError: {e}")
def reset(self):
self.messages = [{"role": "system", "content": self.system_prompt}]
# الاستخدام
if __name__ == "__main__":
bot = SupportBot(model="qwen2.5", temperature=0.4)
print("=== SupportBot V1 ===")
print(bot.chat("Where is my order #12345?"))
print()
print(bot.chat("It has been 7 days since I ordered."))
print()
print(bot.chat("Can I get a refund instead?"))
💻 الإخراج:
TEXT
=== SupportBot V1 ===
I'd be happy to check on your order #12345. Based on our records, your order is currently in transit and expected to arrive within 2-3 business days. You can track it at our website.
I understand your concern. If you'd prefer a refund instead of waiting, I can initiate that for you. Our refund policy covers orders that haven't been delivered within the estimated timeframe.
Yes, I can process a full refund for order #12345. The refund will be credited to your original payment method within 3-5 business days. Would you like me to proceed?
❓ أسئلة شائعة
س فشل pip install ollama، ماذا أفعل؟
ج تأكد من Python ≥ 3.8 وأن pip محدّث:
pip install --upgrade pip. لمشاكل الشبكة، استخدم مرآة: pip install ollama -i https://pypi.tuna.tsinghua.edu.cn/simple.س كيف أختار بين واجهات API المتزامنة وغير المتزامنة؟
ج استخدم المتزامنة للسكربتات والأدوات البسيطة (أبسط). استخدم غير المتزامنة لخدمات الويب (FastAPI/Django) لتجنب حجب حلقة الأحداث. الفرق ضئيل في السيناريوهات أحادية المستخدم.
س الإخراج المتدفق لا يظهر في Jupyter Notebook، ماذا أفعل؟
ج دعم Jupyter لـ
flush=True محدود. استخدم IPython.display.clear_output مع تحديثات الحلقة، أو انتقل للوضع غير المتدفق.س كيف أحدد عنوان خادم Ollama؟
ج الاتصال الافتراضي هو localhost:11434. غيّره عبر المتغير البيئي
OLLAMA_HOST، أو أنشئ Client(host='http://...') في الكود.س ماذا يحدث عندما تصبح قائمة messages طويلة جدًا؟
ج المحتوى الذي يتجاوز نافذة سياق النموذج (num_ctx) يُقتطع. نوصي بتنفيذ نافذة منزلقة، بالاحتفاظ بآخر N أدوار فقط، أو تلخيص المحادثات السابقة.
س كيف أحصل على إحصائيات سرعة الاستنتاج وغيرها؟
ج الاستجابات غير المتدفقة تحتوي على حقول
total_duration وeval_count وprompt_eval_count. آخر قطعة من الاستجابات المتدفقة تحتوي على هذه الإحصائيات.📖 ملخص
- حزمة
ollamaفي Python — تثبيت بسطر واحد، واجهات API متزامنة وغير متزامنة متاحة chat()مناسبة للحوار متعدد الأدوار؛generate()مناسبة للإنشاء أحادي الطلقة- الإخراج المتدفق يُنفَّذ عبر
stream=True+ التكرار على القطع للعرض الفوري - المتدفقة غير المتزامنة تستخدم تكرار
async for، مناسبة للتزامن في خدمات الويب - معالجة الأخطاء يجب أن تغطي ConnectionError وResponseError وTimeoutError
- SupportBot V1 يستخدم dataclass لتغليف سجل المحادثة ومعلمات الاستنتاج
📝 تمارين
- أساسي (صعوبة ⭐): استخدم Python SDK لتنفيذ كل من
chat()وgenerate()مرة واحدة، وقارن فروق الإخراج. - متوسط (صعوبة ⭐⭐): نفّذ دالة محادثة متدفقة بنافذة منزلقة (الاحتفاظ بآخر 5 أدوار)، تتجاهل تلقائيًا أقدم الرسائل عندما تتجاوز المحادثة 5 أدوار.
- متقدم (صعوبة ⭐⭐⭐): ابنِ SupportBot V1+ مع إعادة محاولة الأخطاء والتحكم في المهلة وإخراج بصيغة JSON يحفظ محادثات خدمة العملاء إلى ملف ويسجل وقت الاستنتاج لكل استدعاء.