Spaces:
Sleeping
Sleeping
File size: 5,569 Bytes
a8bc862 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | import re
import datetime
from typing import List, Optional, Annotated, Literal, Dict, Any, Tuple
from enum import Enum
from pydantic import BaseModel, Field, field_validator, model_validator
from pydantic_ai import Agent, RunContext
class CategoryEnum(str, Enum):
ELECTRONICS = 'electronics'
FASHION = 'fashion'
BOOKS = 'books'
HOME_APPLIANCES = 'home_appliances'
OTHERS = 'others'
# class FilterTypeEnum(str, Enum):
# RANGE = 'range'
# MULTISELECT = 'multiselect'
# SINGLESELECT = 'singleselect'
# BOOLEAN = 'boolean'
# class FilterValue(BaseModel):
# type: FilterTypeEnum
# selection: Optional[List[str]]
# range: Optional[Tuple[Optional[float], Optional[float]]]
#
# class UserFilter(BaseModel):
# name: str
# value: FilterValue
class UserFilter(BaseModel):
name: str
type: Literal['range', 'multiselect', 'singleselect', 'boolean']
selection: Optional[List[str]] = None
range: Optional[Tuple[Optional[float], Optional[float]]] = None
class ProductReview(BaseModel):
"""Product Review & perception from other users"""
ratings: Annotated[float, Field(le=5.0, ge=0.0)] = 0.0
num_ratings: int = -1
num_reviews : int = -1
@field_validator('ratings', mode='before')
@classmethod
def val_ratings(cls, x: float | None) -> float:
if not x:
return 0.0
return x
@field_validator('num_ratings', mode='before')
@classmethod
def val_num_ratings(cls, x: str | int) -> int:
if not x:
return -1
if isinstance(x, str):
return int(x.replace(',', ''))
return x
@field_validator('num_reviews', mode='before')
@classmethod
def val_num_reviews(cls, x: str | int) -> int:
if not x:
return -1
if isinstance(x, str):
return int(x.replace(',', ''))
return x
class ProductClass(BaseModel):
"""Category/type of the product desired by the user"""
category: Annotated[CategoryEnum, Field(description="The category of the product")] = CategoryEnum.OTHERS
type: Annotated[str, Field(description="The type of product within the category")] = ''
class Product(BaseModel):
"""Product Specifications"""
id: Optional[str] = None
pro_class : Annotated[Optional[ProductClass] ,Field(description="The category/type of the product")] = None
price: int
name: str
url: str
image: Optional[str]
review: Optional[ProductReview]
details: List[str]
delivery_date: datetime.date | str | None = None
@field_validator('price', mode='before')
@classmethod
def val_price(cls, x: str | int) -> int:
if isinstance(x, str):
return int(re.sub(r'[^\d]', '', x))
# return int(x.replace(',', ''))
return x
class SearchSpecs(BaseModel):
"""Required details for searching the product on the site"""
pro_class : Annotated[ProductClass ,Field(description="The category/type of the product")] = ProductClass()
query: Annotated[str, Field(description="The user's query for the product")] = ''
site: Annotated[str, Field(description="The best website to search for the product")] = ''
site_filters: Annotated[Optional[List[UserFilter]], Field(description="A list of filters available on the site")] = None
filtered_site_filters: Annotated[Optional[List[UserFilter]], Field(description="A list of filters relevant to the user")] = None
user_filters: Annotated[Optional[List[UserFilter]], Field(description="A list of filters to apply when searching for the product")] = None
class ShopResult(BaseModel):
"""Final Result of the shopping agent"""
products: List[Product] = Field(description="A list of products matching the user's query")
recommended: Product = Field(description="The best product matching the user's query")
message: Optional[str] = Field(description="A message indicating if no products were found")
flow: List[str] = []
steps: int = 0
# class SearchResult(BaseModel):
# """Final Result of the shopping agent"""
# products: Annotated[List[Product] ,Field(description="A list of products matching the user's query")]
# best_site: Annotated[str, Field(description="Most suitable web site for shopping the required product")] = ''
# site_filters: Annotated[Optional[List[UserFilter]], Field(description="A list of filters available on the site")] = None
# message: Annotated[Optional[str], Field(description="A message indicating if no products were found")] = 'No products found !'
# class SearchDeps(BaseModel):
# """Dependencies for the base shopping agent"""
# search_specs : SearchSpecs
# og_query: str
# query : str
# best_site: str
class ShopDeps(BaseModel):
"""Dependencies for the base shopping agent"""
og_query: str = ''
query : str
# search_deps : SearchDeps
llm: Any
model_id: str
search_specs : SearchSpecs
# modelConfig: dict = Field(default_factory=lambda: {"temperature":0.2, "max_output_tokens":1024})
modelConfig: Optional[Dict] = None
candidates : Annotated[Optional[List[Product]], Field(description="A list of candidate products")] = None
flow: List[str] = []
steps: int = 0
# playwright states
context_man : Any = None
playwright: Any = None
browser: Any = None
context: Any = None
page: Any = None
@model_validator(mode='after')
def val_query(self):
if not self.query:
self.query = self.og_query
return self
|