نقطة

دليل Pydantic AI 2026: بناء وكلاء LLM آمنين الأنواع بلغة بايثون

AI Bot
بواسطة AI Bot ·
جاري تحميل مشغل تحويل النص إلى كلام الصوتي...

إذا سبق لك أن أطلقت ميزة LLM في بايثون، فأنت تعرف نمط الفشل: يُعيد النموذج نصاً على شكل JSON، تقوم بتحليله، فيتضح أن شيئاً ناقصاً، فتكتب كتلة try/except دفاعية أخرى تخفي المشكلة بدلاً من حلها. Pydantic AI يقلب هذا المنطق رأساً على عقب. فهو يعامل مخططات Pydantic الخاصة بك كعقد يجب على النموذج احترامه، ويوفر فوقها بيئة تشغيل وكلاء بأسلوب FastAPI مع حقن التبعيات — مع البث، والأدوات، وإعادة المحاولة، والمراقبة جاهزة من البداية.

في هذا الدليل، ستثبّت Pydantic AI من الصفر، وتبني وكيل دعم العملاء بمخرجات منظمة وأدوات قاعدة بيانات، وتبدّل بين OpenAI و Anthropic، وتبثّ الرموز إلى نقطة نهاية FastAPI، وتراقب كل شيء باستخدام Logfire لترى ما فعله وكيلك ولماذا.

المتطلبات الأساسية

قبل البدء، تأكد من توفر:

  • بايثون الإصدار 3.10 أو أحدث
  • مفتاح API لمزود واحد على الأقل (OpenAI أو Anthropic أو Google أو Mistral أو نسخة Ollama محلية)
  • معرفة أساسية بنماذج Pydantic و async/await
  • محطة طرفية ومحرر أكواد (يُنصح بـ VS Code مع امتداد Python)
  • اختياري: حساب Logfire للمراقبة (الفئة المجانية كافية تماماً)

ما الذي ستبنيه

في نهاية هذا الدليل، سيكون لديك:

  1. مشروع Pydantic AI ببيئة تبعيات معزولة
  2. وكيل دعم مكتوب بأنواع يُعيد استجابات منظمة محققة بـ Pydantic
  3. أدوات مدعومة بقاعدة بيانات SQLite للبحث في التذاكر وحالة الطلبات
  4. نقطة نهاية محادثة بالبث المباشر معروضة عبر FastAPI
  5. إعداد محايد للمزود يبدّل بين OpenAI و Anthropic و Gemini أثناء التشغيل
  6. مراقبة شاملة مع Logfire، تشمل استخدام الرموز ونطاقات الأدوات
  7. مجموعة اختبارات Pytest تُشغّل الوكيل ضد نموذج LLM وهمي للحصول على اختبارات حتمية وسريعة

الخطوة 1: تثبيت Pydantic AI

يأتي Pydantic AI كحزمة واحدة مع إضافات اختيارية لكل مزود. أنشئ بيئة معزولة أولاً — uv هو الخيار الأسرع في 2026، لكن venv يعمل بنفس الجودة.

mkdir support-agent && cd support-agent
uv venv
source .venv/bin/activate
uv pip install "pydantic-ai[openai,anthropic,logfire]" fastapi uvicorn aiosqlite

إذا كنت تفضل pip التقليدي:

python -m venv .venv
source .venv/bin/activate
pip install "pydantic-ai[openai,anthropic,logfire]" fastapi uvicorn aiosqlite

تحقق من التثبيت:

python -c "import pydantic_ai; print(pydantic_ai.__version__)"

ينبغي أن ترى إصداراً يبدأ بـ 0.x. لا يزال Pydantic AI قبل الإصدار 1.0 وقت كتابة هذا الدليل، لكن واجهة برمجة التطبيقات العامة كانت مستقرة لعدة إصدارات والفريق ملتزم بالتسمية الدلالية للإصدارات القادمة.

الخطوة 2: إعداد المزودين

صدّر مفاتيح API للمزودين الذين تريد استخدامهم. يقرأ Pydantic AI هذه المفاتيح أثناء التشغيل عبر حزم SDK الأساسية، لذلك لا تقم بتسجيلها في Git أبداً.

export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export GEMINI_API_KEY="..."

في المشاريع الحقيقية، ضع هذه المفاتيح في ملف .env وحمّلها باستخدام python-dotenv أو direnv. أضف .env إلى .gitignore فوراً.

الخطوة 3: أول وكيل مكتوب بأنواع

أنشئ ملف agent.py بوكيل بسيط يُعيد SupportResponse منظماً:

# agent.py
from pydantic import BaseModel, Field
from pydantic_ai import Agent
 
 
class SupportResponse(BaseModel):
    """Structured response from the support agent."""
 
    answer: str = Field(description="Plain-language answer for the user")
    needs_human: bool = Field(
        description="True when the issue cannot be resolved by the bot"
    )
    confidence: float = Field(ge=0, le=1, description="Self-rated confidence")
 
 
support_agent = Agent(
    "openai:gpt-4o-mini",
    output_type=SupportResponse,
    system_prompt=(
        "You are a calm, concise customer-support agent for an e-commerce store. "
        "Always return a SupportResponse. Set needs_human=true if the user is angry, "
        "asks for a refund over 200 USD, or mentions legal action."
    ),
)
 
 
if __name__ == "__main__":
    result = support_agent.run_sync(
        "My order #4421 has been stuck on 'shipped' for 14 days. What now?"
    )
    print(result.output)
    print("Tokens used:", result.usage())

شغّل الملف:

python agent.py

يُرسل Pydantic AI موجّه النظام الخاص بك ورسالة المستخدم إلى GPT-4o-mini، ويطلب منه إرجاع JSON يطابق SupportResponse، ويتحقق من الاستجابة باستخدام Pydantic، ويُعطيك كائناً مكتوباً بأنواع كاملة. إذا أعاد النموذج JSON غير صالح أو حقلاً مفقوداً، يُعيد الوكيل المحاولة تلقائياً برسالة تصحيحية — دون الحاجة إلى تحليل يدوي.

الخطوة 4: إضافة حقن التبعيات

تحتاج الوكلاء الحقيقية إلى الوصول إلى قواعد البيانات وعملاء HTTP والإعدادات. يستخدم Pydantic AI نظام تبعيات بأسلوب FastAPI: تُعلن عن deps_type وتصل إليه من داخل الأدوات والموجّهات عبر RunContext.

أنشئ deps.py:

# deps.py
from dataclasses import dataclass
 
import aiosqlite
 
 
@dataclass
class SupportDeps:
    """Resources the agent needs at runtime."""
 
    db: aiosqlite.Connection
    customer_id: int

حدّث agent.py لاستخدامه:

from pydantic_ai import Agent, RunContext
 
from deps import SupportDeps
 
support_agent = Agent(
    "openai:gpt-4o-mini",
    deps_type=SupportDeps,
    output_type=SupportResponse,
    system_prompt=(
        "You are a customer-support agent. Use the provided tools to look up orders "
        "before answering. Never invent order details."
    ),
)
 
 
@support_agent.system_prompt
async def add_customer_context(ctx: RunContext[SupportDeps]) -> str:
    """Inject customer-specific context into every system prompt."""
    async with ctx.deps.db.execute(
        "SELECT name, tier FROM customers WHERE id = ?", (ctx.deps.customer_id,)
    ) as cursor:
        row = await cursor.fetchone()
    if not row:
        return "Customer record not found."
    name, tier = row
    return f"You are speaking with {name} (tier: {tier}). Address them by first name."

لاحظ كيف أن موجّه النظام دالة async عادية مع وصول كامل إلى deps. تحصل على نفس الراحة التي تحصل عليها من تبعية FastAPI، فقط مع كون المستهلك هو نموذج LLM.

الخطوة 5: تعريف الأدوات التي يستطيع النموذج استدعاءها

الأدوات هي دوال async مزينة بـ @support_agent.tool. يفحص Pydantic AI تلميحات الأنواع الخاصة بها ويُولّد مخططات JSON يستطيع النموذج استدعاؤها. تعود القيمة المُرجعة كرسالة أداة في المحادثة.

from datetime import datetime
from typing import Literal
 
 
@support_agent.tool
async def get_order_status(
    ctx: RunContext[SupportDeps],
    order_id: int,
) -> dict:
    """Look up the latest status for a customer order.
 
    Args:
        order_id: The numeric order identifier shown on receipts.
    """
    async with ctx.deps.db.execute(
        "SELECT status, last_update FROM orders WHERE id = ? AND customer_id = ?",
        (order_id, ctx.deps.customer_id),
    ) as cursor:
        row = await cursor.fetchone()
    if not row:
        return {"error": "order_not_found", "order_id": order_id}
    status, last_update = row
    return {
        "order_id": order_id,
        "status": status,
        "last_update": last_update,
        "stale": (datetime.utcnow().isoformat() > last_update),
    }
 
 
@support_agent.tool
async def issue_refund(
    ctx: RunContext[SupportDeps],
    order_id: int,
    amount_cents: int,
    reason: Literal["damaged", "late", "wrong_item", "other"],
) -> dict:
    """Issue a refund for a specific order. Amounts over 20000 cents require human approval."""
    if amount_cents > 20_000:
        return {"approved": False, "reason": "amount_exceeds_bot_limit"}
    await ctx.deps.db.execute(
        "INSERT INTO refunds (order_id, amount_cents, reason) VALUES (?, ?, ?)",
        (order_id, amount_cents, reason),
    )
    await ctx.deps.db.commit()
    return {"approved": True, "order_id": order_id, "amount_cents": amount_cents}

شيئان يجب ملاحظتهما:

  1. تلميحات الأنواع تتحول إلى مخطط. يصبح Literal لـ reason تعداداً يجب على النموذج الاختيار منه. يتحقق Pydantic من كل استدعاء أداة قبل تشغيل دالتك.
  2. الأدوات تستطيع حماية نفسها. ترفض أداة الاسترداد أي مبلغ يزيد على 200 دولار وتدع الوكيل يصعّد الأمر. لا يتعين عليك تعليم النموذج الحد عبر النص — الأداة تفرضه.

الخطوة 6: تشغيل الوكيل مع قاعدة بيانات حقيقية

اربط كل شيء في main.py:

# main.py
import asyncio
 
import aiosqlite
 
from agent import support_agent
from deps import SupportDeps
 
 
async def setup_db() -> aiosqlite.Connection:
    db = await aiosqlite.connect(":memory:")
    await db.executescript(
        """
        CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, tier TEXT);
        CREATE TABLE orders (
          id INTEGER PRIMARY KEY,
          customer_id INTEGER,
          status TEXT,
          last_update TEXT
        );
        CREATE TABLE refunds (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          order_id INTEGER,
          amount_cents INTEGER,
          reason TEXT
        );
        INSERT INTO customers VALUES (1, 'Aya', 'gold');
        INSERT INTO orders VALUES
          (4421, 1, 'shipped', '2026-04-13T08:00:00'),
          (4422, 1, 'delivered', '2026-04-22T14:30:00');
        """
    )
    await db.commit()
    return db
 
 
async def main() -> None:
    db = await setup_db()
    deps = SupportDeps(db=db, customer_id=1)
    result = await support_agent.run(
        "Order #4421 still says shipped after two weeks. Can you refund 35 USD as a goodwill credit?",
        deps=deps,
    )
    print("Answer:", result.output.answer)
    print("Needs human:", result.output.needs_human)
    print("Confidence:", result.output.confidence)
    print("Tool calls:", [m.kind for m in result.all_messages() if m.kind == "tool-call"])
 
 
if __name__ == "__main__":
    asyncio.run(main())

شغّل:

python main.py

سيستدعي الوكيل get_order_status، يرى أن الطلب قديم، يستدعي issue_refund بمبلغ 3500 سنت، ويُعيد SupportResponse يمكنك تمريره مباشرة إلى طبقة واجهة المستخدم.

الخطوة 7: بث الوكيل عبر FastAPI

يكشف Pydantic AI عن agent.run_stream الذي يُعطي مخرجات تدريجية. اقرنه مع StreamingResponse من FastAPI للحصول على نقطة نهاية محادثة تبثّ الرموز إلى المتصفح.

# api.py
from contextlib import asynccontextmanager
 
import aiosqlite
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
 
from agent import support_agent
from deps import SupportDeps
 
 
class ChatBody(BaseModel):
    customer_id: int
    message: str
 
 
db_holder: dict = {}
 
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    db_holder["db"] = await aiosqlite.connect("support.db")
    yield
    await db_holder["db"].close()
 
 
app = FastAPI(lifespan=lifespan)
 
 
@app.post("/chat")
async def chat(body: ChatBody) -> StreamingResponse:
    deps = SupportDeps(db=db_holder["db"], customer_id=body.customer_id)
 
    async def token_stream():
        async with support_agent.run_stream(body.message, deps=deps) as result:
            async for chunk in result.stream_text(delta=True):
                yield chunk
 
    return StreamingResponse(token_stream(), media_type="text/plain")

شغّل الخادم:

uvicorn api:app --reload

ومن محطة طرفية أخرى:

curl -N -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"customer_id": 1, "message": "Where is order 4422?"}'

سترى الرموز تتدفق في الوقت الحقيقي. لأن output_type معيّن، تظل القيمة المجمعة النهائية SupportResponse محققاً — تحصل على تجربة بث دون التخلي عن البنية المنظمة.

الخطوة 8: تبديل المزودين دون إعادة كتابة الكود

الوسيط الأول لـ Agent هو معرّف نموذج. غيّر سلسلة نصية واحدة وسيعمل الوكيل مع مزود مختلف. للبيئات التي يكون فيها الاختيار ديناميكياً، استخدم pydantic_ai.models مباشرة:

import os
 
from pydantic_ai.models.anthropic import AnthropicModel
from pydantic_ai.models.openai import OpenAIModel
 
 
def pick_model():
    name = os.getenv("AGENT_MODEL", "openai:gpt-4o-mini")
    if name.startswith("openai:"):
        return OpenAIModel(name.split(":", 1)[1])
    if name.startswith("anthropic:"):
        return AnthropicModel(name.split(":", 1)[1])
    raise ValueError(f"Unknown model: {name}")
 
 
support_agent = Agent(
    pick_model(),
    deps_type=SupportDeps,
    output_type=SupportResponse,
    system_prompt="...",
)

الآن AGENT_MODEL=anthropic:claude-sonnet-4-6 python main.py يُشغّل نفس الوكيل على Claude Sonnet 4.6 دون أي تغييرات أخرى في الكود. الأدوات والمخرجات المنظمة وحقن التبعيات تعمل جميعاً بشكل متماثل لأن Pydantic AI يُطبّع اختلافات المزودين بدلاً منك.

الخطوة 9: إضافة المراقبة باستخدام Logfire

Pydantic AI مبني من قِبل فريق Pydantic، لذلك يتكامل بشكل أصلي مع Logfire — منتج المراقبة الخاص بهم المبني على OpenTelemetry. كل استدعاء نموذج، واستدعاء أداة، وإعادة محاولة، وخطأ تحقق يصبح نطاقاً يمكنك البحث فيه.

سجّل في logfire.pydantic.dev واحصل على رمز كتابة. ثم:

# observability.py
import logfire
 
logfire.configure(token="your-write-token", service_name="support-agent")
logfire.instrument_pydantic_ai()

استورد هذه الوحدة مرة واحدة في أعلى api.py. أعد تشغيل الخادم وأرسل بضعة طلبات محادثة. في واجهة Logfire سترى:

  • نطاق جذري لكل تشغيل وكيل
  • نطاقات فرعية لكل استدعاء نموذج، مع الموجّه والاستجابة وعدد الرموز
  • نطاقات الأدوات تعرض الوسائط والقيم المُرجعة والمدة
  • نطاقات التحقق عند إعادة محاولة Pydantic AI لمخرجات سيئة

للمراقبة المستضافة ذاتياً، استبدل الاستدعاء بـ logfire.configure(send_to_logfire=False) ووجّه OTLP القياسي إلى مُجمّعك الخاص. الأدوات هي نفسها.

الخطوة 10: اختبار الوكيل دون استهلاك الرموز

يتيح لك pydantic_ai.models.test.TestModel تشغيل اختبارات وكيل شاملة بدون أي اتصال شبكي. يُعيد استجابة منظمة حتمية تطابق output_type الخاص بك، ويمكنك التحقق من استدعاءات الأدوات التي قام بها الوكيل.

# test_agent.py
import pytest
from pydantic_ai.models.test import TestModel
 
from agent import support_agent
from deps import SupportDeps
 
 
@pytest.mark.asyncio
async def test_refund_flow(tmp_db):
    deps = SupportDeps(db=tmp_db, customer_id=1)
    with support_agent.override(model=TestModel()):
        result = await support_agent.run(
            "Refund 50 USD for order 4421",
            deps=deps,
        )
    tool_calls = [m for m in result.all_messages() if m.kind == "tool-call"]
    assert any(t.tool_name == "issue_refund" for t in tool_calls)
    assert isinstance(result.output.answer, str)
    assert 0 <= result.output.confidence <= 1

أضف pytest و pytest-asyncio إلى تبعيات التطوير وشغّل:

pytest -v

تنتهي المجموعة الكاملة في أجزاء من الثانية لأنه لا يُستخدم نموذج LLM حقيقي. استخدم TestModel لاختبارات الوحدة، ثم أضف مجموعة صغيرة من اختبارات التكامل التي تتصل بمزود حقيقي لكل إصدار مرشح.

اختبار التنفيذ الخاص بك

اعبر المسار السعيد بالكامل مرة أخرى:

  1. python main.py يُعيد SupportResponse مع needs_human=False واسترداد مسجل في SQLite
  2. curl على /chat يبثّ الرموز وينتهي بمخرجات منظمة
  3. AGENT_MODEL=anthropic:claude-sonnet-4-6 python main.py يُنتج نفس الشكل على Claude
  4. Logfire يعرض شجرة نطاقات مع استدعاءات الأدوات واستخدام الرموز
  5. pytest يمر في أقل من ثانية باستخدام TestModel

إذا فشل أي من هذه، فالأسباب الأكثر شيوعاً هي مفاتيح API مفقودة، أو إضافة مزود قديمة (شغّل uv pip install --upgrade "pydantic-ai[openai,anthropic]")، أو دالة أداة لا تُعلن أنواعاً يستطيع Pydantic فحصها.

استكشاف الأخطاء وإصلاحها

النموذج يستمر في إرجاع نص حر بدلاً من مخرجات منظمة. تأكد من أن output_type معيّن على Agent وأنك لا تطلب أيضاً نصاً حراً في موجّه النظام. يستخدم Pydantic AI استدعاء الأدوات تحت الغطاء؛ بعض النماذج القديمة تحتاج إلى تثبيتها على متغير قادر على استدعاء الدوال.

أخطاء التحقق تتكرر إلى ما لا نهاية. يُعيد Pydantic AI المحاولة حتى retries=1 بشكل افتراضي. ارفع هذا الرقم بـ Agent(..., retries=3) للنماذج غير المستقرة، لكن إذا كان حقل ما مستحيلاً تحقيقه، فستحرق الرموز. اقرأ خطأ التحقق بعناية — عادة ما يُشير إلى قيد Field صارم جداً.

الأدوات لا تُستدعى أبداً. تحقق من أنك زيّنتها بـ @agent.tool (وليس @agent.tool_plain ما لم ترغب في تخطي RunContext) وأن سلاسل الوثائق الخاصة بها تصف متى يجب استدعاؤها. تعتمد النماذج بشكل كبير على أوصاف الأدوات لاتخاذ القرار.

نقطة نهاية البث تُعيد الرسالة كاملة دفعة واحدة. هذا تخزين مؤقت من FastAPI. تأكد من أنك تُعيد StreamingResponse ولا تنتظر المُولّد قبل الإرجاع.

الخطوات التالية

الخاتمة

يأخذ Pydantic AI الأجزاء التي تعمل بالفعل في تطوير الويب بـ Python — المخططات المكتوبة، وحقن التبعيات، وواجهات برمجة التطبيقات غير المتزامنة أولاً — ويُطبّقها على وكلاء LLM. تتوقف عن التفكير في تحليل JSON والسلاسل النصية بشكل موجّه وتبدأ بالتفكير في العقود: ما الذي يُعيده وكيلي، وما الأدوات التي يستطيع استدعاءها، وما الذي يحتاج إليه للقيام بعمله. النتيجة هي كود وكيل يبدو مثل بقية كود Python الخاص بك، ويُختبر مثل بقية كود Python الخاص بك، ويُشحن بنفس الثقة التي تشحن بها بقية كود Python الخاص بك.

ابنِ بوت الدعم أعلاه، وراقبه بـ Logfire، وفي المرة التالية التي يسألك فيها أحدهم كيف تتعامل مع مخرجات LLM المنظمة، يمكنك الإشارة إلى مجموعة اختبارات ناجحة بدلاً من تعبير منتظم متفائل.

هل تريد قراءة المزيد من الدروس التعليمية؟ تحقق من أحدث درس تعليمي لدينا على استخدام التخزين السحابي لاستراتيجيات النسخ الاحتياطي المحسنة.
العودة إلى الدروس التعليمية

ناقش مشروعك معنا

نحن هنا للمساعدة في احتياجات تطوير الويب الخاصة بك. حدد موعدًا لمناقشة مشروعك وكيف يمكننا مساعدتك.

احجز مكالمة

دعنا نجد أفضل الحلول لاحتياجاتك.

مقالات ذات صلة

بناء وكلاء الذكاء الاصطناعي من الصفر باستخدام TypeScript: إتقان نمط ReAct مع Vercel AI SDK

تعلّم كيفية بناء وكلاء الذكاء الاصطناعي من الأساس باستخدام TypeScript. يغطي هذا الدليل التعليمي نمط ReAct، واستدعاء الأدوات، والاستدلال متعدد الخطوات، وحلقات الوكلاء الجاهزة للإنتاج مع Vercel AI SDK.

35 د قراءة·