Pydantic AI: بناء وكلاء ذكاء اصطناعي آمنة الأنواع مع مخرجات منظمة في Python

pip install 'pydantic-ai[openai]'
# أو لـ Anthropic
pip install 'pydantic-ai[anthropic]'
# أو للجميع
pip install 'pydantic-ai[all]'

uv add 'pydantic-ai[openai]'

export OPENAI_API_KEY="sk-..."
# أو
export ANTHROPIC_API_KEY="sk-ant-..."

from pydantic_ai import Agent
 
agent = Agent(
    'openai:gpt-4o',
    instructions='أنت مساعد مفيد. كن موجزاً ودقيقاً.',
)
 
result = agent.run_sync('ما هي عاصمة تونس؟')
print(result.output)  # "عاصمة تونس هي تونس."

import asyncio
from pydantic_ai import Agent
 
agent = Agent('openai:gpt-4o', instructions='كن موجزاً.')
 
async def main():
    result = await agent.run('اشرح asyncio في جملة واحدة.')
    print(result.output)
 
asyncio.run(main())

# OpenAI
agent = Agent('openai:gpt-4o')
 
# Anthropic
agent = Agent('anthropic:claude-sonnet-4-6')
 
# Gemini
agent = Agent('google-gla:gemini-2.5-flash')
 
# Ollama محلي
agent = Agent('ollama:llama3.2')

from pydantic import BaseModel
from pydantic_ai import Agent
 
class MovieReview(BaseModel):
    title: str
    year: int
    rating: float  # من 0.0 إلى 10.0
    summary: str
    pros: list[str]
    cons: list[str]
    recommended: bool
 
agent = Agent(
    'openai:gpt-4o',
    output_type=MovieReview,
    instructions='أنت ناقد أفلام. أعِد مراجعات أفلام منظمة.',
)
 
result = agent.run_sync('راجع فيلم Inception.')
review = result.output  # النوع: MovieReview — تم التحقق منه بالكامل
 
print(f"{review.title} ({review.year}) — {review.rating}/10")
print(f"موصى به: {review.recommended}")
print("الإيجابيات:", review.pros)

import httpx
from pydantic_ai import Agent, RunContext
 
agent = Agent(
    'openai:gpt-4o',
    instructions='أنت مساعد طقس. استخدم الأدوات لجلب بيانات حقيقية.',
)
 
@agent.tool_plain
async def get_weather(city: str) -> str:
    """جلب حالة الطقس الحالية لمدينة معينة."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f'https://wttr.in/{city}?format=3'
        )
        return response.text
 
async def main():
    result = await agent.run('ما هو الطقس في تونس الآن؟')
    print(result.output)
 
asyncio.run(main())

from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext
 
@dataclass
class ResearchDeps:
    http_client: httpx.AsyncClient
    api_key: str
    max_results: int = 5
 
agent = Agent(
    'anthropic:claude-sonnet-4-6',
    deps_type=ResearchDeps,
    instructions='أنت مساعد بحثي بإمكانيات بحث على الويب.',
)
 
@agent.tool
async def search_web(ctx: RunContext[ResearchDeps], query: str) -> list[str]:
    """البحث في الويب وإعادة قائمة من مقتطفات النتائج ذات الصلة."""
    response = await ctx.deps.http_client.get(
        'https://api.search.example.com/search',
        params={'q': query, 'limit': ctx.deps.max_results},
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
    )
    results = response.json()
    return [r['snippet'] for r in results.get('items', [])]

async def main():
    async with httpx.AsyncClient() as client:
        deps = ResearchDeps(
            http_client=client,
            api_key='your-search-api-key',
            max_results=3,
        )
        result = await agent.run(
            'ابحث في آخر تطورات الحوسبة الكمومية.',
            deps=deps,
        )
        print(result.output)

from pydantic_ai import Agent
 
agent = Agent('openai:gpt-4o', instructions='اكتب محتوى تقنياً مفصلاً.')
 
async def stream_demo():
    async with agent.run_stream('اشرح كيف يعمل TCP/IP.') as stream:
        async for text_chunk in stream.stream_text():
            print(text_chunk, end='', flush=True)
        print()
        result = await stream.get_output()
 
asyncio.run(stream_demo())

from pydantic_ai import Agent
 
agent = Agent('openai:gpt-4o', instructions='أنت مدرّس برمجة.')
 
async def multi_turn():
    # الدور الأول
    result1 = await agent.run('ما هو مزخرف Python؟')
    print('الدور 1:', result1.output)
 
    # الدور الثاني — يمرر الرسائل السابقة كسياق
    result2 = await agent.run(
        'أرني مثالاً عملياً على ذلك.',
        message_history=result1.new_messages(),
    )
    print('الدور 2:', result2.output)
 
asyncio.run(multi_turn())

from pydantic_ai import Agent
from pydantic_ai.usage import UsageLimits
 
agent = Agent('openai:gpt-4o')
 
result = await agent.run(
    'ابحث ولخّص 5 أوراق بحثية في الذكاء الاصطناعي.',
    usage_limits=UsageLimits(
        request_limit=10,        # أقصى أدوار LLM
        tool_calls_limit=20,     # أقصى تنفيذات للأدوات
    ),
)

import asyncio
import httpx
from dataclasses import dataclass
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
 
@dataclass
class Deps:
    client: httpx.AsyncClient
    brave_api_key: str | None = None
 
class ArticleSummary(BaseModel):
    title: str
    url: str
    key_points: list[str] = Field(min_length=1, max_length=5)
 
class ResearchReport(BaseModel):
    topic: str
    executive_summary: str = Field(min_length=50)
    articles: list[ArticleSummary] = Field(min_length=1, max_length=5)
    key_trends: list[str] = Field(min_length=3, max_length=10)
    recommended_actions: list[str] = Field(min_length=1, max_length=5)
    confidence_score: float = Field(ge=0.0, le=1.0)
 
research_agent = Agent(
    'anthropic:claude-sonnet-4-6',
    deps_type=Deps,
    output_type=ResearchReport,
    instructions="""
    أنت محلل بحثي خبير. عند إعطائك موضوعاً:
    1. استخدم أداة البحث للعثور على معلومات حديثة (2-3 عمليات بحث)
    2. استخدم fetch_article للحصول على المحتوى الكامل من أكثر النتائج صلة
    3. اجمع النتائج في ResearchReport منظم
    كن شاملاً وموجزاً. ركّز على الأفكار القابلة للتنفيذ.
    """,
)
 
@research_agent.tool
async def search(ctx: RunContext[Deps], query: str) -> list[dict]:
    """البحث في الويب عن معلومات حديثة حول موضوع ما."""
    if ctx.deps.brave_api_key:
        resp = await ctx.deps.client.get(
            'https://api.search.brave.com/res/v1/web/search',
            headers={'Accept': 'application/json', 'X-Subscription-Token': ctx.deps.brave_api_key},
            params={'q': query, 'count': 5},
        )
        resp.raise_for_status()
        data = resp.json()
        return [
            {'title': r['title'], 'url': r['url'], 'description': r.get('description', '')}
            for r in data.get('web', {}).get('results', [])
        ]
    return [
        {'title': f'مقالة عن {query}', 'url': f'https://example.com/{query}', 'description': f'تغطية شاملة لـ {query}'},
    ]
 
@research_agent.tool
async def fetch_article(ctx: RunContext[Deps], url: str) -> str:
    """جلب وإعادة محتوى نصي لمقالة."""
    try:
        resp = await ctx.deps.client.get(url, timeout=10.0, follow_redirects=True)
        resp.raise_for_status()
        return resp.text[:3000]
    except Exception as e:
        return f'لا يمكن جلب المقالة: {e}'
 
async def research(topic: str) -> ResearchReport:
    async with httpx.AsyncClient(timeout=30.0) as client:
        deps = Deps(client=client)
        result = await research_agent.run(
            f'ابحث في هذا الموضوع بعمق: {topic}',
            deps=deps,
        )
        return result.output
 
async def main():
    report = await research('أطر عمل وكلاء الذكاء الاصطناعي في Python 2026')
    print(f'الموضوع: {report.topic}')
    print(f'درجة الثقة: {report.confidence_score:.0%}')
    print(f'\nالملخص التنفيذي:\n{report.executive_summary}')
    print(f'\nالاتجاهات الرئيسية:')
    for trend in report.key_trends:
        print(f'  - {trend}')
 
if __name__ == '__main__':
    asyncio.run(main())

from pydantic_ai import Agent
from pydantic_ai.models.fallback import FallbackModel
 
# جرّب Claude أولاً، ثم GPT-4o، ثم Gemini
fallback_model = FallbackModel(
    'anthropic:claude-sonnet-4-6',
    'openai:gpt-4o',
    'google-gla:gemini-2.5-flash',
)
 
agent = Agent(fallback_model, output_type=ResearchReport)

import pytest
from pydantic_ai.models.test import TestModel
 
def test_research_agent_structure():
    with research_agent.override(model=TestModel()):
        result = research_agent.run_sync(
            'موضوع اختباري',
            deps=Deps(client=None),
        )
    assert isinstance(result.output, ResearchReport)