Tags:Amazon APIAI AgentLangChainPython电商数据实时数据难度中级 |阅读时长15分钟背景与问题亚马逊大卖家年GMV 1000万的核心竞争优势之一是实时数据能力每15-30分钟采样竞品BSR、价格、库存通过内部BI系统驱动实时决策。这套基础设施的自建成本每年在$60,000-$120,000之间对中小团队完全不可及。本文介绍一套基于Pangolinfo Scrape API LangChain AI Agent的替代架构实现相同数据能力月成本控制在$50以内。系统架构┌─────────────────────────────────────────────────────┐ │ AI Agent Layer │ │ LangChain Agent ← Tools: [BSR监控, 专利扫描, 报告] │ └──────────────────┬──────────────────────────────────┘ │ ┌──────────────────▼──────────────────────────────────┐ │ Data Processing Layer │ │ Python Scripts: 清洗 / 异常检测 / 趋势分析 │ └──────────────────┬──────────────────────────────────┘ │ ┌──────────────────▼──────────────────────────────────┐ │ Data Acquisition Layer │ │ Pangolinfo Scrape API (99% 成功率) │ │ Amazon US/UK/DE/JP Walmart eBay Etsy │ └─────────────────────────────────────────────────────┘环境依赖pipinstallrequests langchain openai pandas schedule python-dotenv# .env 配置PANGOLINFO_API_KEYyour_pangolinfo_key OPENAI_API_KEYyour_openai_key SLACK_WEBHOOK_URLhttps://hooks.slack.com/services/xxx/yyy/zzz核心模块实现1. 数据采集模块# amazon_data_client.pyimportrequestsimportosfromtypingimportOptionalclassAmazonDataClient:Pangolinfo Scrape API 客户端封装BASE_URLhttps://api.pangolinfo.com/v1/amazondef__init__(self,api_key:Optional[str]None):self.api_keyapi_keyoros.getenv(PANGOLINFO_API_KEY)self.sessionrequests.Session()self.session.headers.update({Authorization:fBearer{self.api_key}})defget_product(self,asin:str,marketplace:strUS)-dict:获取商品实时数据BSR、价格、库存、评论respself.session.get(f{self.BASE_URL}/product,params{asin:asin,marketplace:marketplace,fields:bsr,price,inventory_status,review_count,title,brand},timeout15)resp.raise_for_status()returnresp.json()defget_category_top(self,category_id:str,limit:int100)-list:获取品类Top N ASIN列表respself.session.get(f{self.BASE_URL}/category/bestsellers,params{category_id:category_id,marketplace:US,limit:limit},timeout30)resp.raise_for_status()returnresp.json().get(asins,[])defget_search_results(self,keyword:str,pages:int3)-list:获取关键词搜索结果页数据results[]forpageinrange(1,pages1):respself.session.get(f{self.BASE_URL}/search,params{keyword:keyword,marketplace:US,page:page},timeout20)ifresp.ok:results.extend(resp.json().get(results,[]))returnresults2. 异常检测模块# anomaly_detector.pyfromdataclassesimportdataclassfromtypingimportListdataclassclassAlert:alert_type:strasin:strmessage:strseverity:str# HIGH / MEDIUM / LOWaction_suggestion:strclassBSRAnomalyDetector:BSR异常检测断货预警 需求激增检测LOW_STOCK_SIGNALS[Only 1 left in stock,Only 2 left in stock,Only 3 left in stock,Temporarily out of stock,Currently unavailable]def__init__(self,bsr_surge_threshold:int200,window_size:int4):self.bsr_surge_thresholdbsr_surge_threshold self.window_sizewindow_size# 检测窗口采样次数self.bsr_history:dict{}defprocess(self,asin:str,current_bsr:int,inv_status:str)-List[Alert]:alerts[]# 断货预警ifany(sigininv_statusforsiginself.LOW_STOCK_SIGNALS):alerts.append(Alert(alert_typeSTOCKOUT_IMMINENT,asinasin,messagef库存预警:{inv_status},severityHIGH,action_suggestion立即将该品类广告预算提升40-60%窗口期预计12-36小时))# BSR激增检测排名数字下降 销量上升historyself.bsr_history.get(asin,[])iflen(history)2:prev_bsrhistory[-1]bsr_changeprev_bsr-current_bsr# 正值排名上升ifbsr_changeself.bsr_surge_threshold:alerts.append(Alert(alert_typeBSR_SURGE,asinasin,messagefBSR单次上升{bsr_change}位当前排名#{current_bsr},severityMEDIUM,action_suggestion分析需求驱动因素评估是否需要紧急补货))# 更新历史history.append(current_bsr)self.bsr_history[asin]history[-self.window_size:]returnalerts3. AI Agent工具集# agent_tools.pyfromlangchain.toolsimporttoolfromamazon_data_clientimportAmazonDataClientimportjson clientAmazonDataClient()tooldefget_asin_realtime_data(asin:str)-str: 获取指定ASIN的亚马逊实时数据包括BSR排名、价格、库存状态和评论数量。 输入: ASIN字符串例如 B0XXXXXXXXX 输出: JSON格式的实时数据 dataclient.get_product(asin)returnjson.dumps(data,ensure_asciiFalse)tooldefscan_patent_risk(keyword:str)-str: 扫描指定关键词在USPTO/WIPO数据库中的专利风险。 输入: 产品描述关键词英文 输出: 风险等级LOW/MEDIUM/HIGH和相关专利摘要 # 示意实现——实际接入USPTO公开API或专利扫描服务returnjson.dumps({keyword:keyword,risk_level:MEDIUM,note:发现2项相关外观设计专利建议审查设计差异化方案,patents_found:2},ensure_asciiFalse)tooldefget_category_opportunities(category_keyword:str)-str: 分析指定品类关键词的市场机会返回Top20 ASIN的BSR分布和竞争格局。 输入: 品类关键词英文 输出: 市场分析报告JSON resultsclient.get_search_results(category_keyword,pages2)analysis{keyword:category_keyword,total_asins_analyzed:len(results),top_bsr_range:需要实际数据,competition_level:MEDIUM,entry_recommendation:中等竞争建议差异化切入}returnjson.dumps(analysis,ensure_asciiFalse)4. LangChain AI Agent主程序# main_agent.pyfromlangchain.agentsimportinitialize_agent,AgentTypefromlangchain.chat_modelsimportChatOpenAIfromagent_toolsimportget_asin_realtime_data,scan_patent_risk,get_category_opportunitiesimportosdefcreate_product_research_agent():创建亚马逊选品研究AI AgentllmChatOpenAI(modelgpt-4o,temperature0,openai_api_keyos.getenv(OPENAI_API_KEY))tools[get_asin_realtime_data,scan_patent_risk,get_category_opportunities]agentinitialize_agent(toolstools,llmllm,agentAgentType.OPENAI_FUNCTIONS,verboseTrue,max_iterations10)returnagentif__name____main__:agentcreate_product_research_agent()# 示例任务分析POD装饰画品类机会task 分析以下任务 1. 搜索关键词 3D sandstone relief wall art 的品类竞争情况 2. 扫描该品类的专利风险 3. 如果找到候选ASIN获取其实时BSR数据 4. 综合以上数据给出入场建议风险等级、推荐入场价区间、备货时机 resultagent.run(task)print(\n 选品分析报告 )print(result)5. 定时监控调度# scheduler.pyimportscheduleimporttimeimportjsonfromdatetimeimportdatetimefromamazon_data_clientimportAmazonDataClientfromanomaly_detectorimportBSRAnomalyDetectorimportrequestsimportos clientAmazonDataClient()detectorBSRAnomalyDetector(bsr_surge_threshold200)WATCH_LIST[{asin:B0XXXXXXXXX,label:竞品A},{asin:B0YYYYYYYYY,label:竞品B},]defsend_slack_alert(alerts):发送Slack预警通知webhook_urlos.getenv(SLACK_WEBHOOK_URL)ifnotwebhook_urlornotalerts:returnmessage\n.join([f [{a.severity}]{a.alert_type}| ASIN:{a.asin}\nf{a.message}\n{a.action_suggestion}forainalerts])requests.post(webhook_url,json{text:f亚马逊实时监控预警\n{message}})defmonitoring_cycle():单次监控循环print(f[{datetime.now().strftime(%H:%M:%S)}] 开始监控...)all_alerts[]foriteminWATCH_LIST:try:dataclient.get_product(item[asin])alertsdetector.process(asinitem[asin],current_bsrdata.get(bsr,0),inv_statusdata.get(inventory_status,))all_alerts.extend(alerts)exceptExceptionase:print(fError fetching{item[asin]}:{e})ifall_alerts:send_slack_alert(all_alerts)print(f发送{len(all_alerts)}条预警)# 每30分钟执行一次监控schedule.every(30).minutes.do(monitoring_cycle)if__name____main__:print(亚马逊实时数据监控系统已启动30分钟间隔)monitoring_cycle()# 立即执行第一次whileTrue:schedule.run_pending()time.sleep(60)部署建议# 推荐用 supervisor 守护进程[program:amazon_monitor]commandpython /path/to/scheduler.pydirectory/path/to/projectautostarttrueautorestarttruestderr_logfile/var/log/amazon_monitor.err.logstdout_logfile/var/log/amazon_monitor.out.log核心参数调优参考参数保守配置激进配置说明BSR采样间隔60分钟15分钟越短API消耗越大断货预警阈值“Only 3 left”“Only 5 left”越早预警越多误报BSR激增阈值200位100位根据品类BSR分布调整WATCH_LIST大小20-50 ASIN100-200 ASIN影响月API成本如果这套架构对你有帮助欢迎点赞收藏。有技术问题可以在评论区交流或者去Pangolinfo文档中心提工单。
用AI Agent + 亚马逊实时数据API打破大卖家数据垄断：架构设计与完整实现
发布时间：2026/5/19 15:15:51
Tags:Amazon APIAI AgentLangChainPython电商数据实时数据难度中级 |阅读时长15分钟背景与问题亚马逊大卖家年GMV 1000万的核心竞争优势之一是实时数据能力每15-30分钟采样竞品BSR、价格、库存通过内部BI系统驱动实时决策。这套基础设施的自建成本每年在$60,000-$120,000之间对中小团队完全不可及。本文介绍一套基于Pangolinfo Scrape API LangChain AI Agent的替代架构实现相同数据能力月成本控制在$50以内。系统架构┌─────────────────────────────────────────────────────┐ │ AI Agent Layer │ │ LangChain Agent ← Tools: [BSR监控, 专利扫描, 报告] │ └──────────────────┬──────────────────────────────────┘ │ ┌──────────────────▼──────────────────────────────────┐ │ Data Processing Layer │ │ Python Scripts: 清洗 / 异常检测 / 趋势分析 │ └──────────────────┬──────────────────────────────────┘ │ ┌──────────────────▼──────────────────────────────────┐ │ Data Acquisition Layer │ │ Pangolinfo Scrape API (99% 成功率) │ │ Amazon US/UK/DE/JP Walmart eBay Etsy │ └─────────────────────────────────────────────────────┘环境依赖pipinstallrequests langchain openai pandas schedule python-dotenv# .env 配置PANGOLINFO_API_KEYyour_pangolinfo_key OPENAI_API_KEYyour_openai_key SLACK_WEBHOOK_URLhttps://hooks.slack.com/services/xxx/yyy/zzz核心模块实现1. 数据采集模块# amazon_data_client.pyimportrequestsimportosfromtypingimportOptionalclassAmazonDataClient:Pangolinfo Scrape API 客户端封装BASE_URLhttps://api.pangolinfo.com/v1/amazondef__init__(self,api_key:Optional[str]None):self.api_keyapi_keyoros.getenv(PANGOLINFO_API_KEY)self.sessionrequests.Session()self.session.headers.update({Authorization:fBearer{self.api_key}})defget_product(self,asin:str,marketplace:strUS)-dict:获取商品实时数据BSR、价格、库存、评论respself.session.get(f{self.BASE_URL}/product,params{asin:asin,marketplace:marketplace,fields:bsr,price,inventory_status,review_count,title,brand},timeout15)resp.raise_for_status()returnresp.json()defget_category_top(self,category_id:str,limit:int100)-list:获取品类Top N ASIN列表respself.session.get(f{self.BASE_URL}/category/bestsellers,params{category_id:category_id,marketplace:US,limit:limit},timeout30)resp.raise_for_status()returnresp.json().get(asins,[])defget_search_results(self,keyword:str,pages:int3)-list:获取关键词搜索结果页数据results[]forpageinrange(1,pages1):respself.session.get(f{self.BASE_URL}/search,params{keyword:keyword,marketplace:US,page:page},timeout20)ifresp.ok:results.extend(resp.json().get(results,[]))returnresults2. 异常检测模块# anomaly_detector.pyfromdataclassesimportdataclassfromtypingimportListdataclassclassAlert:alert_type:strasin:strmessage:strseverity:str# HIGH / MEDIUM / LOWaction_suggestion:strclassBSRAnomalyDetector:BSR异常检测断货预警 需求激增检测LOW_STOCK_SIGNALS[Only 1 left in stock,Only 2 left in stock,Only 3 left in stock,Temporarily out of stock,Currently unavailable]def__init__(self,bsr_surge_threshold:int200,window_size:int4):self.bsr_surge_thresholdbsr_surge_threshold self.window_sizewindow_size# 检测窗口采样次数self.bsr_history:dict{}defprocess(self,asin:str,current_bsr:int,inv_status:str)-List[Alert]:alerts[]# 断货预警ifany(sigininv_statusforsiginself.LOW_STOCK_SIGNALS):alerts.append(Alert(alert_typeSTOCKOUT_IMMINENT,asinasin,messagef库存预警:{inv_status},severityHIGH,action_suggestion立即将该品类广告预算提升40-60%窗口期预计12-36小时))# BSR激增检测排名数字下降 销量上升historyself.bsr_history.get(asin,[])iflen(history)2:prev_bsrhistory[-1]bsr_changeprev_bsr-current_bsr# 正值排名上升ifbsr_changeself.bsr_surge_threshold:alerts.append(Alert(alert_typeBSR_SURGE,asinasin,messagefBSR单次上升{bsr_change}位当前排名#{current_bsr},severityMEDIUM,action_suggestion分析需求驱动因素评估是否需要紧急补货))# 更新历史history.append(current_bsr)self.bsr_history[asin]history[-self.window_size:]returnalerts3. AI Agent工具集# agent_tools.pyfromlangchain.toolsimporttoolfromamazon_data_clientimportAmazonDataClientimportjson clientAmazonDataClient()tooldefget_asin_realtime_data(asin:str)-str: 获取指定ASIN的亚马逊实时数据包括BSR排名、价格、库存状态和评论数量。 输入: ASIN字符串例如 B0XXXXXXXXX 输出: JSON格式的实时数据 dataclient.get_product(asin)returnjson.dumps(data,ensure_asciiFalse)tooldefscan_patent_risk(keyword:str)-str: 扫描指定关键词在USPTO/WIPO数据库中的专利风险。 输入: 产品描述关键词英文 输出: 风险等级LOW/MEDIUM/HIGH和相关专利摘要 # 示意实现——实际接入USPTO公开API或专利扫描服务returnjson.dumps({keyword:keyword,risk_level:MEDIUM,note:发现2项相关外观设计专利建议审查设计差异化方案,patents_found:2},ensure_asciiFalse)tooldefget_category_opportunities(category_keyword:str)-str: 分析指定品类关键词的市场机会返回Top20 ASIN的BSR分布和竞争格局。 输入: 品类关键词英文 输出: 市场分析报告JSON resultsclient.get_search_results(category_keyword,pages2)analysis{keyword:category_keyword,total_asins_analyzed:len(results),top_bsr_range:需要实际数据,competition_level:MEDIUM,entry_recommendation:中等竞争建议差异化切入}returnjson.dumps(analysis,ensure_asciiFalse)4. LangChain AI Agent主程序# main_agent.pyfromlangchain.agentsimportinitialize_agent,AgentTypefromlangchain.chat_modelsimportChatOpenAIfromagent_toolsimportget_asin_realtime_data,scan_patent_risk,get_category_opportunitiesimportosdefcreate_product_research_agent():创建亚马逊选品研究AI AgentllmChatOpenAI(modelgpt-4o,temperature0,openai_api_keyos.getenv(OPENAI_API_KEY))tools[get_asin_realtime_data,scan_patent_risk,get_category_opportunities]agentinitialize_agent(toolstools,llmllm,agentAgentType.OPENAI_FUNCTIONS,verboseTrue,max_iterations10)returnagentif__name____main__:agentcreate_product_research_agent()# 示例任务分析POD装饰画品类机会task 分析以下任务 1. 搜索关键词 3D sandstone relief wall art 的品类竞争情况 2. 扫描该品类的专利风险 3. 如果找到候选ASIN获取其实时BSR数据 4. 综合以上数据给出入场建议风险等级、推荐入场价区间、备货时机 resultagent.run(task)print(\n 选品分析报告 )print(result)5. 定时监控调度# scheduler.pyimportscheduleimporttimeimportjsonfromdatetimeimportdatetimefromamazon_data_clientimportAmazonDataClientfromanomaly_detectorimportBSRAnomalyDetectorimportrequestsimportos clientAmazonDataClient()detectorBSRAnomalyDetector(bsr_surge_threshold200)WATCH_LIST[{asin:B0XXXXXXXXX,label:竞品A},{asin:B0YYYYYYYYY,label:竞品B},]defsend_slack_alert(alerts):发送Slack预警通知webhook_urlos.getenv(SLACK_WEBHOOK_URL)ifnotwebhook_urlornotalerts:returnmessage\n.join([f [{a.severity}]{a.alert_type}| ASIN:{a.asin}\nf{a.message}\n{a.action_suggestion}forainalerts])requests.post(webhook_url,json{text:f亚马逊实时监控预警\n{message}})defmonitoring_cycle():单次监控循环print(f[{datetime.now().strftime(%H:%M:%S)}] 开始监控...)all_alerts[]foriteminWATCH_LIST:try:dataclient.get_product(item[asin])alertsdetector.process(asinitem[asin],current_bsrdata.get(bsr,0),inv_statusdata.get(inventory_status,))all_alerts.extend(alerts)exceptExceptionase:print(fError fetching{item[asin]}:{e})ifall_alerts:send_slack_alert(all_alerts)print(f发送{len(all_alerts)}条预警)# 每30分钟执行一次监控schedule.every(30).minutes.do(monitoring_cycle)if__name____main__:print(亚马逊实时数据监控系统已启动30分钟间隔)monitoring_cycle()# 立即执行第一次whileTrue:schedule.run_pending()time.sleep(60)部署建议# 推荐用 supervisor 守护进程[program:amazon_monitor]commandpython /path/to/scheduler.pydirectory/path/to/projectautostarttrueautorestarttruestderr_logfile/var/log/amazon_monitor.err.logstdout_logfile/var/log/amazon_monitor.out.log核心参数调优参考参数保守配置激进配置说明BSR采样间隔60分钟15分钟越短API消耗越大断货预警阈值“Only 3 left”“Only 5 left”越早预警越多误报BSR激增阈值200位100位根据品类BSR分布调整WATCH_LIST大小20-50 ASIN100-200 ASIN影响月API成本如果这套架构对你有帮助欢迎点赞收藏。有技术问题可以在评论区交流或者去Pangolinfo文档中心提工单。
高亮度LED恒流驱动控制芯片)
