ترکیب قابلیت vision با فراخوانی توابع

 1!pip install pymupdf --quiet
 2!pip install openai --quiet
 3!pip install matplotlib --quiet
 4!pip install instructor --quiet
 5
 6import base64
 7import os
 8from enum import Enum
 9from io import BytesIO
10from typing import Iterable, List, Literal, Optional
11
12import fitz
13import instructor
14import matplotlib.pyplot as plt
15import pandas as pd
16from IPython.display import display
17from PIL import Image
18from openai import OpenAI
19from pydantic import BaseModel, Field

 1# تابع برای کدگذاری تصویر به صورت base64
 2def encode_image(image_path: str):
 3    if not os.path.exists(image_path):
 4        raise FileNotFoundError(f"Image file not found: {image_path}")
 5    with open(image_path, "rb") as image_file:
 6        return base64.b64encode(image_file.read()).decode('utf-8')
 7
 8# نمونه تصاویر برای تست
 9image_dir = "images"
10image_files = os.listdir(image_dir)
11image_data = {}
12for image_file in image_files:
13    image_path = os.path.join(image_dir, image_file)
14    image_data[image_file.split('.')[0]] = encode_image(image_path)
15    print(f"Encoded image: {image_file}")
16
17def display_images(image_data: dict):
18    fig, axs = plt.subplots(1, 3, figsize=(18, 6))
19    for i, (key, value) in enumerate(image_data.items()):
20        img = Image.open(BytesIO(base64.b64decode(value)))
21        ax = axs[i]
22        ax.imshow(img)
23        ax.axis("off")
24        ax.set_title(key)
25    plt.tight_layout()
26    plt.show()
27
28display_images(image_data)

 1MODEL = "gpt-4-turbo"
 2
 3class Order(BaseModel):
 4    """نمایانگر سفارشی با جزئیات مانند شناسه سفارش، نام مشتری، نام محصول، قیمت، وضعیت و تاریخ تحویل است."""
 5    order_id: str = Field(..., description="شناسه منحصر به فرد سفارش")
 6    product_name: str = Field(..., description="نام محصول")
 7    price: float = Field(..., description="قیمت محصول")
 8    status: str = Field(..., description="وضعیت سفارش")
 9    delivery_date: str = Field(..., description="تاریخ تحویل سفارش")
10
11# توابع جایگزین برای پردازش سفارش
12def get_order_details(order_id):
13    return Order(
14        order_id=order_id,
15        product_name="Product X",
16        price=100.0,
17        status="Delivered",
18        delivery_date="2024-04-10",
19    )
20
21def escalate_to_agent(order: Order, message: str):
22    return f"Order {order.order_id} has been escalated to an agent with message: `{message}`"
23
24def refund_order(order: Order):
25    return f"Order {order.order_id} has been refunded successfully."
26
27def replace_order(order: Order):
28    return f"Order {order.order_id} has been replaced with a new order."
29
30class FunctionCallBase(BaseModel):
31    rationale: Optional[str] = Field(..., description="دلیل برای اقدام.")
32    image_description: Optional[str] = Field(
33        ..., description="توضیحات دقیق تصویر بسته."
34    )
35    action: Literal["escalate_to_agent", "replace_order", "refund_order"]
36    message: Optional[str] = Field(
37        ...,
38        description="پیام به نماینده اگر اقدام به ارجاع به نماینده باشد",
39    )
40
41    def __call__(self, order_id):
42        order: Order = get_order_details(order_id=order_id)
43        if self.action == "escalate_to_agent":
44            return escalate_to_agent(order, self.message)
45        if self.action == "replace_order":
46            return replace_order(order)
47        if self.action == "refund_order":
48            return refund_order(order)
49
50class EscalateToAgent(FunctionCallBase):
51    """ارجاع به نماینده برای کمک بیشتر."""
52    pass
53
54class OrderActionBase(FunctionCallBase):
55    pass
56
57class ReplaceOrder(OrderActionBase):
58    """ابزار برای جایگزینی سفارش."""
59    pass
60
61class RefundOrder(OrderActionBase):
62    """ابزار برای بازپرداخت سفارش."""
63    pass

 1ORDER_ID = "12345" 
 2INSTRUCTION_PROMPT = "You are a customer service assistant for a delivery service, equipped to analyze images of packages. If a package appears damaged in the image, automatically process a refund according to policy. If the package looks wet, initiate a replacement. If the package appears normal and not damaged, escalate to agent. For any other issues or unclear images, escalate to agent. You must always use tools!"
 3
 4def delivery_exception_support_handler(test_image: str):
 5    payload = {
 6        "model": MODEL,
 7        "response_model": Iterable[RefundOrder | ReplaceOrder | EscalateToAgent],
 8        "tool_choice": "auto",  
 9        "temperature": 0.0,  
10        "seed": 123, 
11    }
12    payload["messages"] = [
13        {
14            "role": "user",
15            "content": INSTRUCTION_PROMPT,
16        },
17        {
18            "role": "user",
19            "content": [
20                {
21                    "type": "image_url",
22                    "image_url": {
23                        "url": f"data:image/jpeg;base64,{image_data[test_image]}"
24                    }
25                },
26            ],
27        }
28    ]
29    function_calls = instructor.from_openai(
30        OpenAI(
31            api_key=os.environ.get(("GILAS_API_KEY", "<کلید API خود را اینجا بسازید https://dashboard.gilas.io/apiKey>")), 
32            base_url="https://api.gilas.io/v1/" # Gilas APIs
33        ), mode=instructor.Mode.PARALLEL_TOOLS
34    ).chat.completions.create(**payload)
35    for tool in function_calls:
36        print(f"- Tool call: {tool.action} for provided img: {test_image}")
37        print(f"- Parameters: {tool}")
38        print(f">> Action result: {tool(ORDER_ID)}")
39        return tool
40
41print("Processing delivery exception support for different package images...")
42
43print("\n===================== Simulating user message 1 =====================")
44assert delivery_exception_support_handler("damaged_package").action == "refund_order"
45
46print("\n===================== Simulating user message 2 =====================")
47assert delivery_exception_support_handler("normal_package").action == "escalate_to_agent"
48
49print("\n===================== Simulating user message 3 =====================")
50assert delivery_exception_support_handler("wet_package").action == "replace_order"

 1# تابع برای تبدیل صفحه PDF به تصویر JPEG
 2def convert_pdf_page_to_jpg(pdf_path: str, output_path: str, page_number=0):
 3    if not os.path.exists(pdf_path):
 4        raise FileNotFoundError(f"PDF file not found: {pdf_path}")
 5    doc = fitz.open(pdf_path)
 6    page = doc.load_page(page_number) 
 7    pix = page.get_pixmap()
 8    pix.save(output_path)
 9
10def display_img_local(image_path: str):
11    img = Image.open(image_path)
12    display(img)
13
14pdf_path = 'data/org-chart-sample.pdf'
15output_path = 'org-chart-sample.jpg'
16
17convert_pdf_page_to_jpg(pdf_path, output_path)
18display_img_local(output_path)

 1base64_img = encode_image(output_path)
 2
 3class RoleEnum(str, Enum):
 4    """تعریف نقش‌های ممکن در یک سازمان."""
 5    CEO = "CEO"
 6    CTO = "CTO"
 7    CFO = "CFO"
 8    COO = "COO"
 9    EMPLOYEE = "Employee"
10    MANAGER = "Manager"
11    INTERN = "Intern"
12    OTHER = "Other"
13
14class Employee(BaseModel):
15    employee_name: str = Field(..., description="نام کارمند")
16    role: RoleEnum = Field(..., description="نقش کارمند")
17    manager_name: Optional[str] = Field(None, description="نام مدیر، در صورت وجود")
18    manager_role: Optional[RoleEnum] = Field(None, description="نقش مدیر، در صورت وجود")
19
20class EmployeeList(BaseModel):
21    """لیستی از کارکنان در ساختار سازمانی."""
22    employees: List[Employee] = Field(..., description="لیستی از کارکنان")
23
24def parse_orgchart(base64_img: str) -> EmployeeList:
25    response = instructor.from_openai(OpenAI(
26            api_key=os.environ.get(("GILAS_API_KEY", "<کلید API خود را اینجا بسازید https://dashboard.gilas.io/apiKey>")), 
27            base_url="https://api.gilas.io/v1/" # Gilas APIs
28        )).chat.completions.create(
29        model='gpt-4-turbo',
30        response_model=EmployeeList,
31        messages=[
32            {
33                "role": "user",
34                "content": 'Analyze the given organizational chart and very carefully extract the information.',
35            },
36            {
37                "role": "user",
38                "content": [
39                    {
40                        "type": "image_url",
41                        "image_url": {
42                            "url": f"data:image/jpeg;base64,{base64_img}"
43                        }
44                    },
45                ],
46            }
47        ],
48    )
49    return response
50
51# call the functions to analyze the organizational chart and parse the response
52result = parse_orgchart(base64_img)
53
54# tabulate the extracted data
55df = pd.DataFrame([{
56    'employee_name': employee.employee_name,
57    'role': employee.role.value,
58    'manager_name': employee.manager_name,
59    'manager_role': employee.manager_role.value if employee.manager_role else None
60} for employee in result.employees])
61
62display(df)