@@ -29,6 +29,10 @@ logger = logging.getLogger(__name__)
class AsyncStylistAgent :
ONE_PIECE_SUBS = { " dresses " , " suits " , " jumpsuits " , " bodysuits " , " swimwear " , " underwear " , " lingerie " , " pajamas " }
TOP_SUBS = { " t-shirts " , " shirts " , " blouses " , " polo shirts " , " tank tops " , " sweaters " , " cardigans " , " pullovers " , " hoodies " , " sweatshirts " , " vests " , " coats " , " jackets " , " blazers " , " bras " }
BOTTOM_SUBS = { " trousers " , " pants " , " jeans " , " joggers " , " leggings " , " shorts " , " skirts " , " skorts " }
def __init__ ( self , local_db : str , gemini_model_name : str , outfit_id : str , stylist_name : str , gender : str , callback_url : str ) :
# self.outfit_items: List[Dict[str, str]] = []
self . outfit_id = outfit_id
@@ -56,6 +60,45 @@ class AsyncStylistAgent:
self . minio_bucket = " lanecarford "
self . callback_url = f ' { callback_url } /api/style/callback '
def _is_outfit_complete ( self , occasion : str , first_try : bool = True ) - > bool :
"""
判断当前的 self.outfit_items 是否满足该场合的完整性要求
"""
if occasion not in OCCASION_CATEGORY_MAP :
# 兜底逻辑：如果场合没定义，至少要有衣服和鞋子
raise ValueError ( f " Provided { occasion } is not defined in OCCASION_CATEGORY_MAP. " )
requirements = OCCASION_CATEGORY_MAP [ occasion ]
current_subs = [ item [ ' subcategory ' ] for item in self . outfit_items ]
current_cats = [ item [ ' category ' ] for item in self . outfit_items ]
# --- 1. 鞋子检查 ---
# 如果该场合允许穿鞋（ ） ，
if requirements . get ( " shoes " ) and " shoes " not in current_cats :
return False
# --- 2. 衣服检查 (最核心) ---
# 检查是否有全身单品 (One-piece)
has_one_piece = any ( sub in self . ONE_PIECE_SUBS for sub in current_subs )
if not has_one_piece :
# 如果没有连体衣，必须同时有上装和下装
has_top = any ( sub in self . TOP_SUBS for sub in current_subs )
has_bottom = any ( sub in self . BOTTOM_SUBS for sub in current_subs )
if not ( has_top and has_bottom ) :
return False
# --- 3. 包包检查 ---
# 如果该场合（如 Evening） ， ，
if requirements . get ( " bags " ) and " bags " not in current_cats and first_try :
return False
# --- 4. 配饰检查 ---
if requirements . get ( " accessories " ) and " accessories " not in current_cats and first_try :
return False
return True
def _load_style_guide ( self , stylist_name : str ) :
""" 加载 markdown 风格指南内容。 """
guide_path = os . path . join ( settings . STYLIST_GUIDE_DIR , f " { stylist_name } _en.md " )
@@ -209,6 +252,8 @@ class AsyncStylistAgent:
# 1. 材质偏好 (Occasion Material Map)
material_hint = OCCASION_MATERIAL_MAP . get ( occasion , " " )
# Insert the style_guide content into the template
if occasion == " Bridal / Wedding " :
request_summary + = " IMPORTANT: Strictly only recommend colorful or white items (clothing, shoes, and bags). "
sys_template = template . format (
gender = self . gender ,
current_category = current_category . upper ( ) ,
@@ -422,26 +467,26 @@ class AsyncStylistAgent:
# 4a. 检查类别是否有效 (重要步骤)
if subcategory not in allowed_subcategories :
failed_found_item_count + = 1
logger . warning ( f " Failed to found Item { idx + 1 } : ( { subcategory } ) { rec_item } , invalid subcategory { subcategory } recommended by Agent. " )
continue
# 4b. 在本地 DB 中查询单品
# 4b. 如果这个子类别已经被推荐过了，就舍弃
if subcategory in [ x [ ' subcategory ' ] for x in self . outfit_items ] :
logger . warning ( f " Subcategory { subcategory } already recommended, skipping. " )
continue
# 4c. 在本地 DB 中查询单品
new_item = self . _get_next_item ( description , category , subcategory , occasions , batch_sources , self . gender )
if not new_item or new_item [ ' item_id ' ] in [ x [ ' item_id ' ] for x in self . outfit_items ] :
if not new_item :
failed_found_item_count + = 1
logger . warning ( f " Failed to found Item { idx + 1 } : ( { subcategory } ) { rec_item } , no matching item is found. " )
continue
elif new_item [ ' item_id ' ] in [ x [ ' item_id ' ] for x in self . outfit_items ] :
logger . warning ( f " Found Item { idx + 1 } : ( { subcategory } ) { rec_item } , found item is duplicated. " )
continue
else :
self . outfit_items . append ( new_item )
print ( f " Item { idx + 1 } : ( { subcategory } ) { rec_item } , found item: { new_item } " )
# 如果没有找到的item过于多，
if failed_found_item_count / len ( recommended_items ) > 0.5 :
self . post_operation (
status = " failed " ,
message = f " There are { failed_found_item_count } items (total { len ( recommended_items ) } ) are not found in the database " ,
callback_url = url ,
img_path = merged_image_path
)
raise Exception ( f " There are { failed_found_item_count } items (total { len ( recommended_items ) } ) are not found in the database " )
logger . info ( f " Item { idx + 1 } : ( { subcategory } ) { rec_item } , found item: { new_item } " )
return reason
def _get_allowed_subcategories ( self , occasion : str , category : str ) - > List [ str ] :
@@ -564,11 +609,11 @@ class AsyncStylistAgent:
stylist_guide , accessories_guide = self . _load_style_guide ( self . stylist_name )
url = f ' { callback_url } /api/style/callback '
print ( f " --- Starting Agent (Outfit ID: { self . outfit_id } ) --- " )
MAX_LEN = 9
if not occasions :
occasions = [ " Casual " ]
logger . info ( f """ --- Starting Agent (Outfit ID: { self . outfit_id } ) ---
Occasion: { occasions [ 0 ] } . Stylist: { self . stylist_name } . User ID: { user_id } . Request Summary: { request_summary } . Batch sources: { batch_sources } """ )
general_rules = " \n " . join ( GENERAL_RULES_DICT . values ( ) )
allowed_subcategories = self . _get_allowed_subcategories ( occasions [ 0 ] , " all " )
@@ -582,6 +627,10 @@ class AsyncStylistAgent:
allowed_subcategories ,
MAX_LEN
)
max_retries = 2
for attempt in range ( max_retries ) :
logger . info ( f " Batch recommendation attempt { attempt + 1 } " )
reason = await self . _execute_batch_recommendation (
' all ' , # can be 'accessories' or 'all'
system_prompt ,
@@ -591,6 +640,18 @@ class AsyncStylistAgent:
user_id ,
url
)
if self . _is_outfit_complete ( occasions [ 0 ] , first_try = ( attempt == 0 ) ) :
logger . info ( f " Successfully assembled a complete outfit for { occasions [ 0 ] } after { attempt + 1 } attempts. Reason: { reason } " )
break
else :
self . post_operation (
status = " failed " ,
message = f " Failed to assemble a complete outfit after { max_retries } attempts for { occasions [ 0 ] } . " ,
callback_url = url ,
img_path = " "
)
logger . error ( f " Failed to assemble a complete outfit after { max_retries } attempts for { occasions [ 0 ] } . Current items: { self . outfit_items } . Subcategories required by this occasion is： { allowed_subcategories } " )
raise Exception ( f " Failed to assemble a complete outfit after { max_retries } attempts for { occasions [ 0 ] } . Current items: { self . outfit_items } . Subcategories required by this occasion is： { allowed_subcategories } " )
# 推荐即将完成 回调通知前端
self . post_operation (
pangkaicheng