درس تقني
استخراج JSON منظّم باستخدام OpenAI وPython
أنشئ أداة Python لسطر الأوامر تحمّل ملفات يوميات السفر، وتطلب استخراج بيانات بصيغة JSON من نموذج دردشة من OpenAI، وتتحقق من كل استجابة باستخدام Pydantic، ثم تكتب تقريرًا قابلًا لإعادة الاستخدام.
ما الذي ستبنيه
يبني هذا الدرس travel_journal_analyzer، وهو تطبيق Python صغير لكن مصمم بعقلية إنتاجية. يقبل التطبيق ملفًا نصيًا أو Markdown أو CSV، أو مجلدًا يحتوي على هذه الصيغ. ويحوّل كل مصدر إلى كيان JournalEntry متسق، ثم يرسل الإدخال إلى نموذج دردشة من OpenAI، ويتحقق محليًا من JSON المُعاد، ويكتب تقرير JSON واحدًا للبرمجيات اللاحقة.
للتقرير غرض محدد عمدًا: تحديد المدن المذكورة في الإدخال، والمطاعم المسماة صراحةً، والأطباق، والتقييمات حيثما وردت، والانطباع العام، والنصائح العملية، وملخص موجز. لا يتعامل التطبيق مع بنية JSON الصالحة بوصفها دليلًا على صحة الادعاء. إذ يوجّه الطلب النموذج إلى استخراج الحقائق من الإدخال المزوّد فقط، بينما يتحقق الفحص المحلي من عقد البيانات قبل تصدير النتائج.
يجمع الاستخراج المنظّم الموثوق بين حدّين. الأول هو مخطط JSON المطلوب، الذي يعرّف الحقول التي يتوقعها التطبيق. والثاني هو التحقق المحلي باستخدام Pydantic، الذي يرفض القيم المشوهة، مثل تقييم خارج نطاق خمس نقاط. تصف الأبحاث المتعلقة بالتوليد المنظّم أهمية المخرجات الموثوقة والمحددة الأنواع للتطبيقات التي تحتاج إلى بيانات قابلة للتنبؤ بدلًا من نص نثري غير مقيّد. وفي العمل العملي باستخدام Python، تجعل هذه الحدود الاختبار والصيانة أسهل بكثير.
المتطلبات والإعداد
- Python ٣.١٠ أو إصدار أحدث.
- مفتاح OpenAI API وإمكانية الوصول إلى نموذج دردشة مُعدّ عبر متغير بيئة.
- إلمام أساسي بالطرفية والملفات ودوال Python.
أنشئ مشروعًا وبيئة افتراضية معزولة:
mkdir travel-journal-analyzer
cd travel-journal-analyzer
python -m venv .venv
# macOS or Linux
source .venv/bin/activate
# Windows PowerShell
# .venvScriptsActivate.ps1
python -m pip install --upgrade pip
python -m pip install "openai>=1.0.0" "pydantic>=2.7.0" "python-dotenv>=1.0.1" "pytest>=8.0.0"
mkdir data output testsأنشئ ملف .env. احفظ هذا الملف خارج نظام التحكم بالمصادر. يجب أن تكون مفاتيح API في متغيرات البيئة محليًا، وفي مدير أسرار النشر ضمن بيئة الإنتاج.
OPENAI_API_KEY=your-api-key
OPENAI_MODEL=your-chat-model
MAX_ENTRY_CHARACTERS=12000
REQUEST_TIMEOUT_SECONDS=45أنشئ ملف .gitignore:
.env
.venv/
__pycache__/
.pytest_cache/
output/
*.pycيستخدم هذا الدرس نمط عميل OpenAI الحديث في Python: from openai import OpenAI، ثم client.chat.completions.create(...). لا تستخدم استدعاءات الإكمال القديمة على مستوى الوحدة البرمجية.
الخطوة ١: تعريف عقد البيانات
أنشئ ملف models.py. تمثل الحقول القابلة للقيم الفارغة الحقائق التي لم يوفرها المصدر. وهذا أفضل من اختلاق مدينة أو نوع مطبخ أو تقييم.
from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, Field, field_validator
class JournalEntry(BaseModel):
source_name: str = Field(min_length=1, max_length=255)
entry_id: str = Field(min_length=1, max_length=100)
text: str = Field(min_length=1)
@field_validator("text")
@classmethod
def validate_text(cls, value: str) -> str:
cleaned = value.strip()
if not cleaned:
raise ValueError("Journal entry text cannot be blank.")
return cleaned
class RestaurantFinding(BaseModel):
name: str = Field(min_length=1, max_length=200)
city: str | None = Field(default=None, max_length=120)
country: str | None = Field(default=None, max_length=120)
cuisine: str | None = Field(default=None, max_length=120)
dishes: list[str] = Field(default_factory=list)
rating_out_of_five: float | None = Field(default=None, ge=0, le=5)
sentiment: Literal["positive", "neutral", "negative"]
recommendation_reason: str = Field(min_length=1, max_length=600)
class JournalAnalysis(BaseModel):
entry_id: str = Field(min_length=1, max_length=100)
cities_mentioned: list[str] = Field(default_factory=list)
restaurants: list[RestaurantFinding] = Field(default_factory=list)
travel_tips: list[str] = Field(default_factory=list)
concise_summary: str = Field(min_length=1, max_length=1000)
class AnalysisReport(BaseModel):
generated_at_utc: str
model: str
total_entries: int = Field(ge=0)
successful_analyses: int = Field(ge=0)
failed_entries: list[str] = Field(default_factory=list)
analyses: list[JournalAnalysis] = Field(default_factory=list)هذه النماذج ليست مجرد توثيق. إذ تحوّل JournalAnalysis.model_validate_json() استجابة النموذج إلى كائن تم التحقق منه. وتفشل الاستجابة التي تحتوي على تسمية انطباع غير صالحة أو تقييم من ست نقاط قبل أن تصل إلى قاعدة بيانات أو جدول بيانات أو واجهة موجهة للعملاء.
الخطوة ٢: تحميل المدخلات النصية وCSV
أنشئ ملف journal_loader.py. تنشئ ملفات النص العادي وMarkdown إدخالًا واحدًا لكل ملف. ويتطلب ملف CSV عمود text وينشئ إدخالًا واحدًا لكل صف غير فارغ. يمنع حد الأحرف الطلبات الكبيرة على نحو غير متوقع.
from __future__ import annotationsimport csv
from pathlib import Path
from models import JournalEntrySUPPORTED_SUFFIXES = {".txt", ".md", ".csv"}def load_journal_entries(path_value: str, max_characters: int) -> list[JournalEntry]:
path = Path(path_value).expanduser().resolve()
if not path.exists():
raise FileNotFoundError(f"Input path does not exist: {path}")if path.is_dir():
entries: list[JournalEntry] = []
for child in sorted(path.iterdir()):
if child.is_file() and child.suffix.lower() in SUPPORTED_SUFFIXES:
entries.extend(load_journal_entries(str(child), max_characters))
if not entries:
raise ValueError("Directory contains no supported input files.")
return entriesif path.suffix.lower() in {".txt", ".md"}:
text = path.read_text(encoding="utf-8").strip()
_check_length(text, path.name, max_characters)
return [JournalEntry(source_name=path.name, entry_id=path.stem, text=text)]if path.suffix.lower() != ".csv":
raise ValueError("Use a .txt, .md, .csv file, or directory.")entries = []
with path.open("r", encoding="utf-8-sig", newline="") as handle:
reader = csv.DictReader(handle)
if not reader.fieldnames or "text" not in reader.fieldnames:
raise ValueError("CSV must contain a column named 'text'.")
for row_number, row in enumerate(reader, start=2):
...تابع القراءة
سجل دخولك مجاناً لقراءة المقال كاملاً والوصول إلى أدوات الذكاء الاصطناعي.
تسجيل الدخول / إنشاء حساب