用GPTBots构建亚马逊竞品动态跟踪 Agent，双引擎架构实现实时市场洞察

摘要：在竞争激烈的亚马逊电商市场中，实时的竞品动态跟踪是品牌成功的关键。然而，传统的竞品分析工具往往只能提供静态数据，难以满足快速变化的市场需求。本文将介绍如何利用GPTBots构建一个智能的竞品动态跟踪Agent，通过“历史学家”和“侦察兵”的双引擎架构，实现对亚

在竞争激烈的亚马逊电商市场中，实时的竞品动态跟踪是品牌成功的关键。然而，传统的竞品分析工具往往只能提供静态数据，难以满足快速变化的市场需求。本文将介绍如何利用GPTBots构建一个智能的竞品动态跟踪Agent，通过“历史学家”和“侦察兵”的双引擎架构，实现对亚马逊竞品的实时洞察和动态分析。

你是否也厌倦了那些刚做完就“过时”的竞品分析报告？

在亚马逊这个瞬息万变的战场，静态的Excel表格和月度PPT早已跟不上节奏。当我们还在分析上个月的榜单时，新的爆款可能已经悄然崛起，旧的王者或许已因一次差评而跌落神坛。

我们需要一个能实时对话、能按需深挖、能理解我们复杂商业问题的智能系统。

之前，我探索过用AI工具链自动化内容生产，解决了从0到1的问题。但这次，挑战升级了。

我需要一个真正的“竞品动态追踪Agent” ，它不仅要能回答“上周谁是第一？”，更要能回答“最近上榜的新品里，主打‘轻量化’卖点的有哪些？它们的营销做得怎么样？”

要实现这个目标，我们需要一个能够将“离线数据沉淀”和“实时动态交互” 结合的平台。

这就是我再次选择GPTBots的原因——它作为一个企业级的LLMOps平台，提供了我们搭建这种复杂系统所需的一切“重武器”：数据库、可编排工作流、以及一个足够聪明的智能体大脑。

感兴趣可以注册体验：https://www.gptbots.AI/signup?utm_source=kol=binggan

接下来，饼干哥哥带你一步步复现，如何用GPTBots打造一个“活”的竞品追踪系统。

在开始构建这个复杂的系统之前，我们先来定义一个具体、真实的战场。

我们是一家专注于智能宠物用品的公司，我们的下一个目标，就是切入竞争激烈的“智能猫砂盆”市场。

我们每天都需要紧盯着亚马逊上这个类目的Best Seller榜单，链接在此： https://www.amazon.com/Best-Sellers-Pet-Supplies-Self-Cleaning-Cat-Litter-Boxes/zgbs/pet-supplies/17602464011/ref=zg_bs_nav_pet-supplies_4_2975299011

但一张静态的截图能告诉我们什么？它无法回答这些至关重要的商业问题：昨天排名第三的那个产品，今天还在榜上吗？那个新上榜的#2产品，是什么时候上架的？它的用户口碑如何？榜单上的这些产品，哪些在A+页面和视频营销上投入了更多资源？

这些问题，便是我们构建之旅的起点。

一、以终为始：设计一个“历史学家”+“侦察兵”的双引擎系统

在动手之前，我们先画出蓝图。这个系统的核心，是两个协同工作的角色：1. 历史学家（离线工作流）：它是一个不知疲倦的记录员，通过一个独立的、定时运行的工作流，每天自动抓取亚马逊Best Seller榜单，并将带有时间戳的排名数据，精准地存入GPTBots的数据库。它负责构建我们所有分析的基石——一部完整的竞品历史档案。2. 前线侦察兵（实时智能体）：这是我们直接交互的对象。它是一个聪明的分析师，大脑里装着整个历史数据库。当你提出问题时，它不仅能快速查询历史，更能根据你的复杂需求，动态地、逐一地派出一个“侦察兵”（另一个实时工作流），去抓取任何一个竞品当下的详细页面信息，然后将所有情报汇总，给你一个有深度的答案。

这个“历史+实时”的双引擎架构，就是我们整个系统的灵魂。

整体框架如mermaid 示意图：

二、动手开干：组装三大核心组件

明确了目标，我们就可以在GPTBots里开始“施工”了。整个搭建过程，主要分为三大块：数据库、离线工作流、和实时工作流。数据库

根据前面说的，这个系统的中心是智能体，所以要先搭建好。

然后在智能体里，新建好数据表，用于存储历史的静态排名数据

要先定义字段

可以通过AI对话添加的形式处理：表格用于记录亚马逊的 best sellers 商品排名，需要字段：排名、名称、链接，以及记录插入的日期。

检查数据无误后，添加至字段列表即可。

数据表格新建好了，接下来就要考虑怎么把数据存进去。

✅ 这里需要通过 API 的形式

我在文档中找到了说明：

其中 table_id是如图这里：

但要想通过 API 存数据，要先把 Agent 发布，然后在集成中，打开 API

然后创建API Key

OK，准备好了。

这个工作流的目标是：定时运行，抓取榜单，存入数据库。

在GPTBots里，这个过程被拆解得非常清晰：第一步：获取页面数据。利用能稳定抓取亚马逊的工具（例如你用的Scrapeless API）获取Best Seller页面的原始信息。第二步：LLM数据解析。将原始信息喂给一个LLM节点，通过一个精准的提示词，让它提取出我们想要的30个商品的排名、标题和链接，并格式化为标准的JSON数组。 [图片：展示LLM节点解析数据的提示词和输出结果的截图]第三步：代码入库。这是最关键的一步。我们使用一个代码节点，编写Python脚本。这个脚本接收上一步生成的json，然后调用GPTBots的数据库API，将这30条数据，连同当前的日期，一同写入我们的 amazon_best_sellers 数据库。

第一步是用 jinja 获取页面数据

没有任何配置可言，直接给网址就行了

返回的数据长这样（节选）：{ “code”: 200, “data”: { “images”: { “Image 35: KITPLUS Self Cleaning Litter Box, Automatic Cat Litter Box with APP Control, Odor-Free Waste Disposal Includes Trash Bags, La”: “https://images-na.ssl-images-amazon.com/images/I/61sLBGNUCML._AC_UL600_SR600,400_.jpg”, ….nodeId=468496&ref_=footer_privacy)n* [Consumer Health Data Privacy Disclosure](https://www.amazon.com/gp/help/customer/display.html?ie=UTF8&nodeId=TnACMrGVghHocjL8KB&ref_=footer_consumer_health_data_privacy)n* [Your Ads Privacy Choices](https://www.amazon.com/privacyprefs?ref_=footer_iba)nn© 1996-2025, Amazon.com, Inc. or its affiliates” }, “meta”: { “usage”: { “tokens”: 14723 } }, “status”: 20000}

接着配置一个 LLM 从其中抓取需要的数据，转成 JSON

其中的Prompt：你的作用是一个数据提取器，从用户输入的内容中，提取出产品商品的排名、标题与链接，并以以下数组 array 格式返回：[{“rank”:”1″,”title”:”标题 1″,”website”:”网址 1″},{“rank”:”2″,”title”:”标题 2″,”website”:”网址 2″},{“rank”:”3″,”title”:”标题 3″,”链website”:”网址 3″}]注意用户输入的是从亚马逊 best seller 排行榜页面抓取的 json 内容，需要从中解析出 30 个商品信息和链接。而排名则是根据商品出现的顺序。确保输出的是标准的 array 格式，后续可以直接通过 python 的 eval 或者 json.loads 解析。其余的什么都不要说。

测试一下，确保无误。

输出的JSON 结构数据长这样：{ “数据”:[ { “排名”:”1″, “标题”:”Cumrige Self Cleaning Litter Box, Large Capacity Automatic Cat Litter Box Self Cleaning for Cats, App Control,Safety Protection, 2 Roll Garbage Bags,White & Grey”, “链接”:”https://www.amazon.com/Cumrige-Cleaning-Automatic-Multiple-Protection/dp/B0DFM4Z71X/ref=zg_bs_g_17602464011_d_sccl_1/132-8327553-5397737?psc=1″ }, { “排名”:”2″, “标题”:”Self Cleaning Cat Litter Box, Automatic Cat Litter Box for Multiple Cats, Integrated Safety Protection with Open-Top Design”, “链接”:”https://www.amazon.com/Cleaning-Automatic-Multiple-Integrated-Protection/dp/B0DQW7857Q/ref=zg_bs_g_17602464011_d_sccl_2/132-8327553-5397737?psc=1″ }, … { “排名”:”30″, “标题”:”Automatic Cat Litter Box Self Cleaning for Multiple Cats – XL Smart Litter Box with App Control, Odor-Free Waste Disposal, Quiet Operation, Safety Sensors, Compatible with Clumping Litter”, “链接”:”https://www.amazon.com/Automatic-Litter-Self-Cleaning-Multiple/dp/B0F9S3SP21/ref=zg_bs_g_17602464011_d_sccl_30/132-8327553-5397737?psc=1″ }]}

接着是代码节点，直接把 json 数据存入数据库

首先，不着急马上存入，逐步跑通。

先测一下LLM 输出的是否为标准的 json，可以在 python 被解析成字典：def main(arg: str) -> dict: # 从输入参数中获取文本字符串 dict_ = eval(arg) # 将提取的数据作为名为 “result” 的变量输出 return { “result”: dict_ }

OK，没问题。

此时再补充代码，根据开头新建数据库后，拿到的表 id 和 key，就可以同步了。import Requestsimport jsonfrom typing importList, Dict, Anyfrom datetime import datetimedefmain(arg: str) -> dict: “”” 接收一个包含产品列表的JSON字符串，解析后将其同步到指定的数据库表。 Args: arg: 一个JSON格式的字符串，例如 ‘[{“rank”:”1″,”title”:”标题1″,”website”:”网址1″}]’ Returns: 一个包含操作结果的字典。 “”” # — 请在这里配置您的 API 信息 — # 请在 API 密钥页面获取密钥作为 token api_key = “app-s865Iwhif82YsebTz2sI4c6M” # 请替换为您的数据表 ID table_id = “687ca423bf62602706f8cdaf” # api_url = “https://api.gptbots.ai/v1/database/import/records” headers = { “Authorization”: f”Bearer {api_key}”, “Content-Type”: “application/json” } try: # 步骤 1: 解析输入的字符串为 Python 列表 # 使用 json.loads 比 eval 更安全 data_list = json.loads(arg) ifnotisinstance(data_list, list): raise ValueError(“输入的数据不是一个列表”) except (json.JSONDecodeError, ValueError) as e: print(f”输入参数解析失败: {e}”) return {“status”: “Failed”, “message”: “输入参数格式错误，无法解析。”} # 步骤 2: 构建符合 API 要求的数据记录 (records) records = # 获取当前时间，并格式化为字符串 current_time = datetime.now.strftime(‘%Y-%m-%d %H:%M:%S’) for item in data_list: # 注意：API文档中的字段是 “website”，但数据库截图中的是 “url” # 这里我们假设数据库截图中的 “url” 是正确的字段名 record = { “values”: { “rank”: item.get(“rank”), “title”: item.get(“title”), “url”: item.get(“website”), # 从输入的 ‘website’ 映射到数据库的 ‘url’ “inserted_date”: current_time } } records.append(record) # 步骤 3: 构建最终的请求体 (payload) payload = { “table_id”: table_id, “records”: records } # 步骤 4: 发送 API 请求 try: response = requests.post(api_url, headers=headers, json=payload, timeout=30) response.raise_for_status # 如果请求失败 (状态码 4xx 或 5xx), 会抛出 HTTPError response_data = response.json # 根据API文档，检查返回的 code 是否为 0 if response_data.get(“code”) == 0: print(f”成功导入 {len(records)} 条数据。”) return {“status”: “Success”, “data”: response_data} else: print(f”API返回错误: {response_data.get(‘message’)}”) return {“status”: “Failed”, “message”: response_data.get(“message”)} except requests.exceptions.RequestException as e: print(f”请求失败: {e}”) return {“status”: “Failed”, “message”: f”请求API时发生网络错误: {e}”} except Exception as e: print(f”未知错误: {e}”) return {“status”: “Failed”, “message”: f”发生未知错误: {e}”}

顺利同步到数据库了。

构建实时单品抓取工作流

这个工作流非常纯粹，它的任务只有一个：接收一个产品URL，返回该页面的详细信息。

这里有个问题，就是亚马逊的接口非常容易触发反爬，我用 jinja 等工具都翻车了。

最后还是用上次我们介绍的工具——Scrapeless才能解决，他有专门抓取亚马逊的接口：

如何注册使用见我这文章：我用n8n+Scrapeless搭了个「SEO内容引擎」，全自动挖掘高转化选题，解决出海流量难题

这个工作流我们将其命名为 scrape_amazon，使用Scrapeless的亚马逊产品接口，确保能稳定获取数据，然后通过LLM节点，将其解析成我们需要的、包含所有丰富维度的JSON格式。

老规矩，直接创建工作流，要注意描述要填写清楚，Agent才能正确识别并调用：

代码节点：调用 Scrapeless 从亚马逊单品页面获得数据

import jsonimport timefrom dataclasses import dataclassimport requests@dataclassclassPayload: actor: str input: dictdefget_task_result(host: str, task_id: str, token: str): url = f”https://{host}/api/v1/scraper/result/{task_id}” headers = {“x-api-token”: token} return requests.get(url, headers=headers)defsend_request(amazon_url): host = “api.scrapeless.com” url = f”https://{host}/api/v1/scraper/request” token = “sk_…填入你的 token” headers = { “Content-Type”: “application/json”, “x-api-token”: token } input_data = { “action”: “product”, “domain”: “www.amazon.com”, “url”: amazon_url } payload = Payload(actor=”scraper.amazon”, input=input_data) try: response = requests.post(url, headers=headers, data=json.dumps(payload.__dict__)) print(“response:”, response.text) if response.status_code == 200: print(“[✅] Request sent successfully.”) return response.text task_id = response.json.get(“taskId”) ifnot task_id: print(“[!] No taskId found in response.”) returnNone print(f”[*] Task submitted successfully. Task ID: {task_id}”) except Exception as e: print(f”[!] Error sending scrape request: {e}”) returnNone # Polling for result whileTrue: time.sleep(1) result_data = get_task_result(host, task_id, token) print(f”[*] Checking result for Task ID: {task_id}…”) if result_data.status_code == 200: return result_data.text defmain(arg_1: str) -> dict: result = send_request(arg_1) return { “result”: result }

LLM 节点：从结果中解析出必要的 JSON 数据

核心提示词：## Role你是一个高度精准的Web数据解析与结构化引擎，专门处理来自Scrapeless的亚马逊产品JSON数据。## Task你的核心任务是：接收一份复杂的原始JSON输入，并严格遵循【目标JSON结构与指令】，将其转换为一个信息丰富、结构清晰的JSON对象，用于后续的商业智能分析。—## 目标JSON结构你必须生成以下结构的单一JSON对象。{“rank”:null,”asin”:null,”title”:null,”brand”:null,”manufacturer”:null,”rating”:null,”reviews_count”:null,”date_first_available”:null,”image_url”:null,”video_count”:null,”has_a_plus_content”:null,”categories_json”:null,”item_weight”:null,”product_dimensions”:null,”top_review_text”:null,”customer_says_summary”:null,”product_specifics_json”:null,”price_info”:null}

万事俱备，只欠一个“大脑”——我们的竞品洞察Agent。

在GPTBots中新建智能体，然后将我们创建好的数据库 amazon_best_sellers 和工作流 scrape_amazon 挂载上去。

最核心的，是它的身份提示词（System Prompt）。一个好的提示词，是教会AI如何思考。我放弃了简单的策略罗列，而是为它设计了一个“复合链式调用” 的思维模式。## Role你是一个顶级的亚马逊市场洞察专家，部署在 gptbots.ai 平台。你的核心能力在于深度理解用户意图，并动态地、链式地调用数据库和工作流工具，以提供精准的、数据驱动的竞品分析。## 可用资源1.**数据库 (DB)**: `amazon_best_sellers` — 存储了每日亚马逊Best Seller榜单的历史快照数据。2.**工作流 (Workflow)**: `scrape_amazon(product_url)` — 输入一个亚马逊产品URL，可以实时抓取该产品详情页的丰富信息。—## 核心思维模式与执行策略你的工作流程不是固定的，而是根据用户问题的复杂度，在以下两种策略中动态选择和组合。### 策略1: 直接事实查询 (单一工具调用)**适用场景**: 当用户的问题非常直接，可以通过单次查询数据库就能回答时。* **示例**: * “查一下2025年7月20日，ASIN为B0BQXHK363的产品排名是多少？” * “上周五’Home & Kitchen’类目下，排名前5的产品有哪些？”* **执行步骤**: 1. 理解用户意图，识别出查询的实体（ASIN、日期、类目）。 2. 直接查询数据库 `amazon_best_sellers`。 3. 基于查询结果，直接回答用户。### 策略2: 复合查询与深度分析 (复合链式调用)**适用场景**: 当用户的问题是复杂的、探索性的，需要结合历史趋势和产品实时详情才能回答时。**这是你区别于简单查询工具的关键能力。*** **示例**: * “最近一周新上榜的竞品中，有没有主打‘轻量化’(item_weight)这个卖点的？” * “帮我分析一下本周排名上升最快的5个产品，它们的A+页面（has_a_plus_content）和视频（video_count）营销做得怎么样？” * “我想找找Top 20里，有哪些是最近3个月内上架的新品，并且用户口碑（rating, customer_says_summary）不错的？”* **执行步骤 (你必须严格遵循此逻辑链条)**: 1. **第一步：理解与拆解 (Understand & Decompose)** * 首先，将用户的复杂问题拆解为几个子任务。 * **例如**: 对于问题“查找主打‘轻量化’的新品”，你会拆解为： * 任务A：定义并找出“最近一周新上榜的竞品”列表。 * 任务B：获取这个列表中每个产品的详细信息，特别是`item_weight`。 * 任务C：整合信息，筛选出符合“轻量化”标准的产品并总结汇报。 2.**第二步：数据库查询 (Database Query)** * 执行**任务A**。要找出“新上榜”或“排名上升”的产品，你必须进行**比较查询**。这意味着你需要查询**至少两个时间点**的数据（例如，本周和上周），然后在内部进行对比，找出差异化的产品列表（主要是ASIN列表）。 3.**第三步：迭代调用工作流 (Iterative Workflow Calls)** * 这是**最关键的一步**。执行**任务B**。 * 获取到第二步的ASIN列表后，你**必须对列表中的每一个ASIN**，构建其亚马逊产品URL，并**逐个、循环地调用工作流 `scrape_amazon`**，以获取它们详细、实时的产品信息。 4. **第四步：综合分析与回答 (Synthesize & Answer)** * 执行**任务C**。将所有工作流返回的详细信息进行汇总。 * 基于汇总后的信息，进行最终的筛选、分析和判断（例如，判断`item_weight`是否属于“轻量化”范畴）。 * 最后，用清晰、有条理的语言向用户呈现你的分析结论，而不仅仅是罗列数据。—## 约束* **严禁幻觉**: 你不知道的信息就是不知道，不能编造。你的所有回答都必须基于可用工具返回的数据。* **链式调用是核心**: 面对任何探索性、分析性的问题，你都必须启动【策略2】的思维模式，即“先查数据库获得列表，再循环调用工作流获取详情”。* **透明化过程**: 在执行复杂查询（策略2）时，可以向用户简要说明你的执行计划，例如：“好的，我将首先查询数据库找出近期上榜的新品列表，然后逐一分析它们的产品详情。请稍候。” 这样可以提升用户体验。

这个提示词的核心思想是：1. 识别问题类型：当用户提问时，先判断这是个简单的事实查询，还是一个复杂的探索性问题。2. 拆解复杂问题：如果问题复杂（比如“找主打‘轻量化’的新品”），Agent必须先在内部把它拆解成多个步骤：任务A: 先去数据库找出“新品”的列表。任务B: 对这个列表里的每一个产品，去调用 scrape_amazon 工作流获取详情。任务C: 汇总所有详情，进行分析和筛选，最后回答。3. 链式调用工具：这套“先查库、再循环调工作流”的逻辑，就是AI解决复杂问题的关键。它将两个独立的工具，串联成了一条强大的分析流水线。

三、使用

配置完成后，我们来问几个问题，看看这个Agent的真实能力。

先小试牛刀问个简单的问题：抓取 top1 的产品信息

Agent 能正确到数据库找到排第一的数据，并且调用Scrape_amazon 的工作流来爬取数据：

最终给出信息总结：

接下来，再问一个复杂的：“最近上榜的新品中，有没有主打‘轻量化’（item_weight）这个卖点的？”

这正是考验我们“链式调用”设计的时刻。Agent的表现堪称完美：1. 首先，它向我说明了它的分析计划：“好的，我会分两步为您分析：1. 查询数据库，找出最近上榜的新品。2. 获取这些新品的详细信息，重点关注它们的商品重量…”2. 接着，它执行了第一步，查询数据库，并返回了它找到的新品ASIN列表。 3. 然后，它自动对列表中的产品执行了第二步，逐个调用 scrape_amazon 工作流，在后台获取了所有产品的详细重量等信息。4. 最后，它汇总所有信息，给出了一个综合性的结论：“该新品虽然是近期上榜，但其重量为24磅，属于中大型产品，并不主打‘轻量化’卖点…”

这背后的逻辑：1. 先在数据库中找新上榜的产品

2. 把新上榜的多个产品链接，逐个调用工作流scrape_amazon 读取信息，用于回答问题

四、升级到FlowAgent

前面我们用单个智能体实现的“链式调用”，说实话，是常规的玩法。它能解决问题，但就像一个全能但劳累的特种兵，面对更复杂的任务时，还是会显得力不从心。

也就是说，让单个Agent支持复杂的多步骤任务，其实不是最佳解法，FlowAgent才是。

为什么要升级到FlowAgent？

简单来说，当我们的任务逻辑从“如果A，就做B” 升级到 “先判断是A还是B，如果是A就走流程1，如果是B就先查C再查D，最后汇总给E” 这种复杂的分支与并行处理时，FlowAgent就是不二之选。

我们的“竞品动态追踪”场景，恰好就是一个典型的复杂业务场景：简单的提问：“昨天Top 1是谁？”——直接查数据库就行。复杂的问题：“最近上榜的5个新品，哪个A+页面做得最好？”——这就需要先查库找出5个URL，然后并行地、同时派5个“侦察兵”（工作流）去抓取页面信息，最后再汇总分析。

用单个Agent去循环调用5次工作流，是串行的，慢；

而用FlowAgent，可以实现并行处理，快！这就是效率的代差。

2. 我们来创建一个FlowAgent

最大的不同就在于，智能体的设置里不再有那个长长的“身份提示词（System Prompt）”，取而代之的是一个可视化的、可以拖拽编排的 Flow 画布。

我们要搭建的Flow是这样的，别看这图上节点很多，显得眼花缭乱，但其实逻辑非常简单，拆解开来就一目了然：

整个Flow的核心逻辑分为五个部分：1. 第一级分类器 (Classifier)：智能路由，判断用户意图。2. LLM-数据库查询：处理简单的、基于历史数据的查询。3. 第二级分类器 (Classifier)：URL分发器，将数据库查出的多个URL分发给不同的处理单元。4. 并行的LLMs：多个LLM同时启动，分别调用scrape_amazon工作流处理不同的URL。5. LLM-总结回答：汇总所有并行处理的结果，形成最终的、综合性的答案。

3. 动手开干：组装FlowAgent的核心组件

第一步：入口-智能分类器

这是整个Flow的大脑中枢。它接收用户的原始问题，并作出第一个关键判断：这个问题是想查具体的竞品信息，还是想问其他问题？

第二步：LLM-数据库查询模块

这个LLM节点连接着分支一。它的任务很简单：理解用户问题，去数据库amazon_best_sellers里查找对应的产品URL。为了能让后续节点处理，我们通过提示词，强制它以固定的JSON格式输出最多5个URL。

Prompt：## 任务原则必须向用户简要说明你的执行计划，以提升用户体验。例如：“好的，我将首先查询数据库找出近期上榜的新品列表，然后逐一分析它们的产品详情。## Role你是一个顶级的亚马逊市场洞察专家，善于使用各类工具完成用户任务。## 任务执行规则1. 数据库表amazon_best_sellers存储了每日亚马逊 Best Seller 榜单竞品产品名称、URL等历史快照数据。当查询到结果后，输出URL列表## 约束- **严禁幻觉**: 你不知道的信息就是不知道，不能编造。你的所有回答都必须基于可用工具返回的数据。- 若未查询数据库到5个产品URL，则仅输出所查询到的数量## 输出必须按照以下JSON格式输出：{“URL1”: “”},{“URL2”: “”},{“URL3”: “”},{“URL4”: “”},{“URL5”: “”}

第三步：URL分发器（还是分类器）

这个节点是实现并行处理的关键。它接收上一个LLM节点输出的JSON，然后将里面的URL1, URL2, URL3… 分发到不同的路径上。

配置非常简单，为每个URL创建一个分支即可。

第四步：连接并行的LLMs（协同军团）

这是FlowAgent最酷炫的地方。我们创建5个一模一样的LLM节点，分别连接到URL分发器的5个出口。

这5个LLM会同时接收到各自的URL，然后同时调用 scrape_amazon 工作流去抓取信息。

## 任务原则必须向用户简要说明你的执行计划，以提升用户体验。例如：“好的，我将首先查询数据库找出近期上榜的新品列表，然后逐一分析它们的产品详情。## Role你是一个顶级的亚马逊市场洞察专家，善于使用各类工具完成用户任务，输出产品分析结论## 任务执行规则1. 调用crape_amazon输入一个亚马逊产品URL，可以实时抓取该产品介绍、价格等详情页的丰富信息。2.你的工作流程不是固定的，而是根据用户问题的复杂度，如果需要，可以重复多次调用工具以完成任务。- 基于数据，用清晰、有条理的语言向用户呈现你的分析结论。## 约束- **严禁幻觉**: 你不知道的信息就是不知道，不能编造。你的所有回答都必须基于可用工具返回的数据。- **“调用amazon_best_sellers“的前提是必须有一个正确的亚马逊产品URL，否则禁止调用

第五步：收口-最终的总结LLM

所有“侦察兵”完成任务后，我们需要一个“指挥官”来汇总战报。在Flow的末端，我们添加最后一个LLM节点，并将前面5个并行LLM的输出全部连接到它的输入。

它的任务就是将所有零散的分析结果，整合成一段通顺、连贯、有洞察的最终回答。