完成AI提示词分析竞品的定义

docs/gpt/agent_product.md

@@ -0,0 +1,22 @@
+# Agent prompt template
+@/src\ai\agent_product.py 
+@/src\models\product_model.py 
+帮我修改 task 方法，我主要用于提取有关数据给 llm 作为提示词，我需要的数据有 crawl_data.extra_result.product_info crawl_data.extra_result.result_table 不过为了节省字符串长度，有些字段是不需要的，例如 image_url 、 goto_amazon、 img_path 、keyword_link、amazon_search_link 、unique_words
+
+并且在筛选完成后，这些字段理论上应该会有描述信息，现在我只能在字段串中硬编码的定义，是否有更加通用和优雅的办法获取自动获得该模型的字段名和描述词。不想硬编码 analyz_main_keyword_template_str 
+```python
+    analyz_main_keyword_template_str = '''
+我是亚马逊运营，我在给产品名称为 {product_name} 选主要关键词，附件中是我从同类竞品的关键词搜索量数据，总共有 {competitor_count} 个竞品数据。
+例如 
+- asin 是竞品商品的编号
+- product_info 是该竞品商品的信息
+- result_table.traffic_keyword 是该竞品商品的同类相关的关键词搜索量数据，
+- monthly_searches 是同类相关的关键词月搜索量
+它所在的列是它的关键词，第二列是该竞品搜索关键词月搜索量。
+末尾包含了该产品的所有信息。
+往右是第二个商B0CQ1SHD8V 也是一样。帮我选出他们两个的相同关键词并且搜索量在1万以上来作为我产品的主要关键词3个。
+如果竞品的所有关键词搜索量都没有达到1万以上的话，就从排名前十的关键词里筛选三个搜索量最大相关性最强的词最为主关键词。
+'''
+    text_qa_template = PromptTemplate(analyz_main_keyword_template_str)
+
+```

src/ai/agent_product.py

@@ -1,44 +1,138 @@
+import json
+from typing import Optional
 from llama_index.core import PromptTemplate
-
 import asyncio
 import aiofiles
 import os
 import sys
-from src.models.product_model import Product,CompetitorCrawlData,AICompetitorAnalyzeMainKeywords, SearchAmazoneKeyResult, ProductBaseInfo, Variant
+from pydantic import BaseModel
+from src.models.product_model import (
+    Product, CompetitorCrawlData, AICompetitorAnalyzeMainKeywords,
+    TrafficKeywordResult, ProductImageInfo,
+    SearchAmazoneKeyResult, ProductBaseInfo, Variant
+)
 from src.manager.core.db_mongo import BaseMongoManager
 from utils.logu import get_logger
+from src.models.field_config import FieldConfig
+
 logger = get_logger('ai')
-async def task():
-    db_mongo = BaseMongoManager()
-    await db_mongo.initialize()
-    product = await Product.find_one(Product.basic_info.name == "电线保护套")
+
+# 默认包含的字段配置
+DEFAULT_FIELD_CONFIG = FieldConfig(
+    include_fields={
+        "ProductImageInfo": {
+            "main_text"  # 产品图片主要文字
+        },
+        "TrafficKeywordResult": {
+            "traffic_keyword",  # 流量关键词名称
+            "monthly_searches"  # 关键词月搜索量
+        },
+        "ProductBaseInfo": {
+            "name", "content", "material", "color", "size",
+            "packaging_size", "weight", "main_usage", "selling_point"
+        },
+        "CompetitorCrawlData": {
+            "asin",
+        }
+    }
+)
+
+def get_competitor_prompt_data(
+    product: Product,
+    field_config: FieldConfig = DEFAULT_FIELD_CONFIG
+) -> list:
+    """
+    获取竞品提示数据
+    
+    Args:
+        product: 产品对象
+        field_config: 字段配置
+        
+    Returns:
+        结构化竞品数据列表
+    """
     competitor_crawl_data = product.competitor_crawl_data
-    for asin,crawl_data in competitor_crawl_data.items():
-        logger.info(f"{asin}")
-        logger.info(f"{crawl_data.extra_result.product_info}")
-        logger.info(f"{crawl_data.extra_result.result_table}")
-        break
-    return
+    list_data = []
+    
+    for asin, crawl_data in competitor_crawl_data.items():
+        if crawl_data.extra_result:
+            structured_result = {"asin": asin}
+            
+            if crawl_data.extra_result.product_info:
+                structured_result["product_info"] = field_config.filter_model_dump(
+                    crawl_data.extra_result.product_info,
+                    "ProductImageInfo"
+                )
+                
+            if crawl_data.extra_result.result_table:
+                structured_result["result_table"] = [
+                    field_config.filter_model_dump(item, "TrafficKeywordResult")
+                    for item in crawl_data.extra_result.result_table
+                ]
+                
+            logger.debug(f"Structured result for LLM: {json.dumps(structured_result, indent=4, ensure_ascii=False)}")
+            list_data.append(structured_result)
+            
+    return list_data
+
+def get_field_descriptions(
+    model_class: BaseModel,
+    field_config: FieldConfig = DEFAULT_FIELD_CONFIG,
+    model_name: Optional[str] = None
+) -> dict:
+    """
+    获取模型字段描述信息
+    
+    Args:
+        model_class: Pydantic模型类
+        field_config: 字段配置
+        model_name: 模型名称(用于查找配置)
+        
+    Returns:
+        字段名到描述的字典
+    """
+    return field_config.get_model_fields(model_class, model_name)
 
 async def test_product_mongo():
     db_mongo = BaseMongoManager()
     await db_mongo.initialize()
     product = await Product.find_one(Product.basic_info.name == "电线保护套")
-    for crawl_data in product.competitor_crawl_data.values():
-        logger.info(f"{crawl_data.extra_result}")
-        break
+    product_name = product.basic_info.name
+    # 使用默认配置
+    competitor_data = get_competitor_prompt_data(product)
+    competitor_desc = get_field_descriptions(CompetitorCrawlData)
+    product_info_desc = get_field_descriptions(ProductImageInfo)
+    keyword_result_desc = get_field_descriptions(TrafficKeywordResult)
+    
+    output_format = [{"asin": "", "main_key":"", "monthly_searches":"", "reason":""}]
+    # logger.info(f"competitor_data {competitor_data}")
+    logger.info(f"competitor_desc {competitor_desc}")
+    logger.info(f"product_info_desc {product_info_desc}")
+    logger.info(f"keyword_result_desc {keyword_result_desc}")
+    
     analyz_main_keyword_template_str = '''
-    我是亚马逊运营，我在给产品名称为数据线保护套选主要关键词，附件中是我从同类竞品的关键词搜索量数据。
-    例如 B0B658JC22 是第一个商品的竞品数据，它所在的列是它的关键词，第二列是该竞品搜索关键词月搜索量。
-    末尾包含了该产品的所有信息。
-    往右是第二个商B0CQ1SHD8V 也是一样。帮我选出他们两个的相同关键词并且搜索量在1万以上来作为我产品的主要关键词3个。
-    如果竞品的所有关键词搜索量都没有达到1万以上的话，就刷选，就从排名前十的关键词里筛选三个搜索量最大相关性最强的词最为主关键词。
-    '''
-    text_qa_template = PromptTemplate(analyz_main_keyword_template_str)
-
+各个字段说明：
+{desc}
 
+竞品数据：
+{competitor_data}
+----
+我是亚马逊运营，我在给产品名称为 {product_name} 选主要关键词，以上数据是我从同类竞品的关键词搜索量数据，总共有 {competitor_count} 个竞品数据。
+请帮我分析这些竞品数据，选出搜索量在1万以上的相同关键词作为主要关键词3个。
+如果竞品的搜索量都不足1万，则从排名前十的关键词中筛选三个搜索量最大且相关性最强的词。
+输出格式：
+{output_format}
+'''
+    text_qa_template = PromptTemplate(analyz_main_keyword_template_str)
+    logger.info(f"{text_qa_template.format(
+        desc=(competitor_desc, product_info_desc, keyword_result_desc),
+        product_name=product_name,
+        competitor_data=competitor_data,
+        competitor_count=len(competitor_data),
+        output_format=output_format,
+)}")
 def main():
-    asyncio.run(task())
+    asyncio.run(test_product_mongo())
 
 if __name__ == "__main__":
     main()

src/models/field_config.py

@@ -0,0 +1,77 @@
+from typing import Dict, List, Optional, Set
+from pydantic import BaseModel
+
+class FieldConfig:
+    """字段配置管理器（仅包含模式）"""
+    
+    def __init__(
+        self,
+        include_fields: Dict[str, Set[str]],
+        field_descriptions: Optional[Dict[str, Dict[str, str]]] = None
+    ):
+        """
+        初始化字段配置
+        
+        Args:
+            include_fields: 包含的字段配置，格式为{"模型类名": {"字段1", "字段2"}}
+            field_descriptions: 字段描述配置，格式为{"模型类名": {"字段1": "描述1", "字段2": "描述2"}}
+        """
+        self.include_fields = include_fields
+        self.field_descriptions = field_descriptions or {}
+
+    def get_model_fields(
+        self,
+        model_class: BaseModel,
+        model_name: Optional[str] = None
+    ) -> Dict[str, str]:
+        """
+        获取模型的字段配置
+        
+        Args:
+            model_class: Pydantic模型类
+            model_name: 模型名称(用于查找配置)
+        
+        Returns:
+            字段名到描述的字典
+        """
+        model_name = model_name or model_class.__name__
+        include = self.include_fields.get(model_name, set())
+        
+        field_descriptions = {}
+        for field_name, field_info in model_class.model_fields.items():
+            # 只返回包含列表中指定的字段
+            if field_name in include:
+                # 优先使用配置中的描述，其次使用模型中的描述
+                description = (
+                    self.field_descriptions.get(model_name, {}).get(field_name)
+                    or field_info.description
+                )
+                if description:
+                    field_descriptions[field_name] = description
+                
+        return field_descriptions
+
+    def filter_model_dump(
+        self,
+        model_instance: BaseModel,
+        model_name: Optional[str] = None
+    ) -> Dict:
+        """
+        过滤模型dump结果
+        
+        Args:
+            model_instance: 模型实例
+            model_name: 模型名称(用于查找配置)
+        
+        Returns:
+            过滤后的字典
+        """
+        model_name = model_name or model_instance.__class__.__name__
+        include = self.include_fields.get(model_name, set())
+        
+        dump_data = model_instance.model_dump()
+        return {
+            field_name: value 
+            for field_name, value in dump_data.items()
+            if field_name in include
+        }

src/models/product_model.py

@@ -16,19 +16,19 @@ class MarketingInfo(BaseModel):
 
 class ProductBaseInfo(BaseModel):
     """产品基本信息
-产品名称	电线保护套			
-包装内容	一个/袋			
-材质	TPU			
-颜色	三种颜色：黑色、银色、金色			
-尺寸	直径6MM，长度93-95CM			
-包裹尺寸	14*17*3CM			
-重量	30G			
-主要用途	保护电源线，车匝线，刹车线			
-主要卖点	
-- 优质TPU，健康环保好看，耐磨耐脏			
-- 耐高温防火阻燃，			
-- 预防线管老化，延长线路使用寿命			
-- 柔软易弯曲，可卷起来收纳			
+产品名称	电线保护套
+包装内容	一个/袋
+材质	TPU
+颜色	三种颜色：黑色、银色、金色
+尺寸	直径6MM，长度93-95CM
+包裹尺寸	14*17*3CM
+重量	30G
+主要用途	保护电源线，车匝线，刹车线
+主要卖点
+- 优质TPU，健康环保好看，耐磨耐脏
+- 耐高温防火阻燃，
+- 预防线管老化，延长线路使用寿命
+- 柔软易弯曲，可卷起来收纳
     """
     name: Optional[str] = None
     content: Optional[str] = None
@@ -52,17 +52,41 @@ class Variant(BaseModel):
 
 class TrafficKeywordResult(BaseModel):
     """流量关键词结果"""
-    traffic_keyword: Optional[str] = None
-    keyword_link: Optional[str] = None
-    monthly_searches: Optional[str] = None
-    amazon_search_link: Optional[str] = None
+    traffic_keyword: Optional[str] = Field(
+        default=None,
+        description="竞品同类相似的关键词名称"
+    )
+    keyword_link: Optional[str] = Field(
+        default=None,
+        description="关键词详情页链接"
+    )
+    monthly_searches: Optional[str] = Field(
+        default=None,
+        description="竞品同类相似关键词的月搜索量"
+    )
+    amazon_search_link: Optional[str] = Field(
+        default=None,
+        description="亚马逊搜索链接"
+    )
 
 class ProductImageInfo(BaseModel):
     """产品图片信息"""
-    image_url: Optional[str] = None
-    goto_amazon: Optional[str] = None
-    main_text: Optional[str] = None
-    img_path: Optional[str] = None
+    image_url: Optional[str] = Field(
+        default=None,
+        description="产品图片URL"
+    )
+    goto_amazon: Optional[str] = Field(
+        default=None,
+        description="跳转亚马逊链接"
+    )
+    main_text: Optional[str] = Field(
+        default=None,
+        description="该 asin 商品的主要文字内容"
+    )
+    img_path: Optional[str] = Field(
+        default=None,
+        description="本地图片存储路径"
+    )
 
 class ExtraResultModel(BaseModel):
     """额外结果数据模型"""
@@ -73,7 +97,7 @@ class ExtraResultModel(BaseModel):
 class CompetitorCrawlData(BaseModel):
     '''竞品表'''
     sql_id: Optional[int] = Field(default=None)
-    asin: Optional[str] = None
+    asin: Optional[str] = Field(default=None, description="竞品商品的编号")
     asin_area: Optional[str] = 'JP'
     # 爬取数据的 S3 路径
     extra_result_path: Optional[str] = None
@@ -123,7 +147,7 @@ class Product(Document):
         description="竞品分析信息，使用JSONB存储。竞品主关键词分析、竞品长尾词分析。。。")
     competitor_analyze: Optional[List[AICompetitorAnalyzeMainKeywords]] = Field(
         default=CompetitorAnalyze(),
-        description="竞品分析信息，使用JSONB存储。竞品主关键词分析、竞品长尾词分析。。。" 
+        description="" 
     )
     # 变体，如版本、型号、颜色、套餐，各个变体对应着价格、成本等财务数据
     variants: Optional[List[Variant]] = Field(