本文是《AI Agent 实战》系列第 7 篇也是第一个完整业务实战。前六篇我们练完了所有单项技能工具调用第 1 篇、任务规划第 2 篇、知识库第 3 篇、调试第 4 篇、接渠道第 5 篇、安全确认第 6 篇。这一篇把它们组装成一个真的能帮人干活的东西你丢给它一个关键词它去合规数据源拉候选商品、算指标、打分、让大模型做软评最后推一张确认卡片给你——你点确认它才生成选品报告。系列代码基于 Python 3.11 LangGraph 1.x数据侧以淘宝联盟开放平台接口为主线示例。所有平台接口权限、额度、收费与法规条文均为 2026 年 10 月检索口径请以官方页面为准。一、先当买手再当工程师选品的业务流程拆解很多做选品 Agent 的人第一反应是写个爬虫抓淘宝这是本末倒置。先搞清楚人类买手怎么选品Agent 只是把这套动作自动化。一个成熟的电商运营做一次选品大致走五步定方向拿到宠物智能用品这种模糊需求后先拓展出十几个候选关键词圈定类目和价格带捞候选在货源平台1688和零售平台淘宝/天猫分别搜索人工翻几十页把看起来能做的先存下来算账对每个候选算三笔账——拿货价和零售价之间毛利够不够、同款卖的人多不多竞争度、最近卖得动不动增速看细节对少数入围款人肉去看差评集中在什么问题上、供应商靠不靠谱、有没有知识产权风险拍板老板/运营负责人做最终决策——这个环节人从来不让机器代劳。对应到工程上就是一个五层架构意图层把自然语言需求变成结构化搜索条件采集层从合规数据源拉原始数据数据层清洗并计算指标决策层硬打分加 LLM 软评交付层出报告中间插一道第 6 篇讲的人工确认闸口。这套分层不是花架子——它决定了你调试时问题出在哪一层。数据层算错毛利率你在决策层怎么换模型都救不回来。第 4 篇的老话先定位层再定位节点最后定位行。给新手一个判断标准五层里只有 L4 的一半允许出现大模型。意图解析、软评、报告文案可以是 LLM采集、清洗、打分、交付必须是确定性代码。凡是每次跑结果应该一样的环节交给模型都是在给未来的自己埋随机性炸弹——你会在复盘时永远分不清选品选错了还是模型今天心情不好。二、数据从哪来2026 年的合规红线与三条可行路这一节是全篇最重要的没有之一。选品 Agent 的技术难度是中等但法律风险是高——因为你的核心原料商品数据长在别人的平台上。先说红线。2025 年 10 月 15 日起施行的新修订《反不正当竞争法》新增了被称为数据专款的第十三条第三款经营者不得以欺诈、胁迫、避开或者破坏技术管理措施等不正当方式获取、使用其他经营者合法持有的数据。此前最高法已在 2025 年 8 月发布数据权益专题指导性案例宁波中院也判过突破反爬措施抓取淘宝天猫商品数据并有偿披露的构成不正当竞争情节更重的突破登录态、控制他人系统还可能触及《刑法》第 285 条非法获取计算机信息系统数据罪。落到实操层面的翻译是禁止绕验证码/登录墙/反爬机制抓页面、伪造 UA 高频打接口、抓用户个人信息评价人昵称手机号等禁止把抓来的数据打包卖或公开再分发可以走平台官方开放接口在授权范围内做自用分析。还有个容易被忽略的点第三方数据接口服务要慎选。市面上大量号称淘宝全字段 API的聚合服务上游来源不明你付了钱可能买到的是别人违规爬取的数据下游风险会顺着采购链条传导给你。判断方法很简单对方说不出数据授权来源的一律不用。官方接口有三条路按接入成本从低到高路一淘宝联盟阿里妈妈物料搜索 API。对个人开发者最友好的一条路。taobao.tbk.dg.material.optional.upgrade物料搜索升级版属于免费接口、不需要用户授权返回商品标题、价格、销量、佣金率、优惠券等选品核心字段。门槛是注册淘宝开放平台账号、实名认证、创建应用拿到 AppKey/AppSecret并在淘宝联盟后台申请推广位拿到 PID格式mm_xxx_xxx_xxx三者必须匹配。个人实名认证账号可查基础商品数据但日额度低普遍口径在每日数百到一千次量级、部分高级字段拿不到商用需企业主体。具体权限与额度以开放平台控制台和官方文档为准。路二1688 开放平台。选品最终要找货源1688 的alibaba.item.search/alibaba.item.get能拿到批发价、起订量MOQ、供应商诚信通年限等 B 端字段是算毛利的主力数据源。但必须企业开发者主体企业支付宝认证个人号调不了核心接口计费是基础免费资源包按量三层免费层搜索 QPS 约 10 次/秒资源包年费在数百到三千元区间。以 open.1688.com 现行规则为准。路三手动导出 CSV。生意参谋、联盟后台都支持人工导出数据报表。写一个只吃本地 CSV 的采集适配器第一天就让它跑通全链路——这是我最推荐的起步方式零合规风险、零审核等待先把 Agent 逻辑调对再换真接口。本篇正文代码用路一做主示例、路三做兜底演示。三条路怎么切换也有讲究采集层做成统一接口search(q, filters) - list[dict]联盟 API、1688 API、CSV 是三个实现类跑哪个由配置决定。这样审核期用 CSV 不阻塞开发接口下来改一行配置切换数据层往上完全无感。架构上这叫端口-适配器在选品这种数据源永远在变的业务里是保命设计。另外提醒一句拿到的数据只在你自己的分析闭环里用别做二次分发别碰用户个人信息这是第 20 篇会展开讲的内容但从今天第一篇实战开始就要守。三、采集层TOP 签名、物料搜索封装与限流淘宝开放平台TOP所有接口共用一套签名规则公共参数 业务参数剔除sign和空值后按参数名 ASCII 升序拼成key1value1key2value2...前后包上 AppSecret 做 MD5或 HMAC-MD5结果转大写。时间戳误差不能超过 15 分钟。# collect_tbk.py —— 淘宝联盟物料搜索封装Python 3.11 import time, hmac, hashlib import requests from dataclasses import dataclass GATEWAY https://eco.taobao.com/router/rest def top_sign(params: dict, secret: str, method: str md5) - str: TOP 签名ASCII 升序拼接 - 前后包 secret - MD5/HMAC-MD5 - 大写 items {k: v for k, v in params.items() if k ! sign and v not in (, None)} base .join(f{k}{items[k]} for k in sorted(items)) if method md5: return hashlib.md5(f{secret}{base}{secret}.encode()).hexdigest().upper() return hmac.new(secret.encode(), base.encode(), hashlib.md5).hexdigest().upper() dataclass class TbkClient: app_key: str app_secret: str adzone_id: str # 联盟后台申请的推广位 PID 中的 adzoneId def search(self, q: str, *, page: int 1, size: int 40, price_lo: float | None None, price_hi: float | None None, sort: str total_sales_des) - list[dict]: params { app_key: self.app_key, method: taobao.tbk.dg.material.optional.upgrade, v: 2.0, format: json, sign_method: md5, timestamp: time.strftime(%Y-%m-%d %H:%M:%S), adzone_id: self.adzone_id, q: q, page_no: str(page), page_size: str(size), sort: sort, } if price_lo is not None: params[start_price] str(price_lo) if price_hi is not None: params[end_price] str(price_hi) params[sign] top_sign(params, self.app_secret) r requests.post(GATEWAY, dataparams, timeout10) r.raise_for_status() data r.json() if error_response in data: raise RuntimeError(fTOP 接口错误: {data[error_response]}) tl data[tbk_dg_material_optional_upgrade_response][result_list][map_data] return tl class Throttle: 令牌桶限流个人号额度低主动限速比被 ISP_FLOW_CONTROL_LIMIT 打回体面 def __init__(self, min_interval: float 2.0): self.min_interval, self._last min_interval, 0.0 def wait(self): gap time.time() - self._last if gap self.min_interval: time.sleep(self.min_interval - gap) self._last time.time()三个容易踩的坑提前说start_tk_rate这类比率参数是万分位1234 12.34%page_size上限 100但个人号建议 20-40返回结构里result_list.map_data在只有一条数据时可能退化成 dict 而不是 list解析前必须做类型归一。CSV 兜底适配器长这样接口审核没下来之前先用它开发# collect_csv.py —— 手动导出报表兜底字段名与物料搜索对齐 import csv def load_csv(path: str) - list[dict]: rows [] with open(path, encodingutf-8-sig) as f: for r in csv.DictReader(f): rows.append({ num_iid: r[商品ID], title: r[标题], zk_final_price: r[券后价], uv: r.get(近30天销量, 0), commission_rate: r.get(佣金率, 0), source: csv, }) return rows再交代两个工程细节。其一密钥管理AppSecret 和 DashScope Key 一样走环境变量或.env python-dotenv永远不要写进代码——这是要提交进 Git 的文件泄一次就得去控制台重置。其二沙箱先行TOP 提供沙箱环境和测试数据签名逻辑、错误分支先在沙箱调顺再切正式网关正式环境的日额度是稀缺资源别拿它当调试器用一次联调烧掉半天额度是新手最常见的浪费。四、数据层清洗、去重与三个核心指标原始数据进来先过三道清洗标题去 emoji 与促销噪声词包邮拍下备注、按num_iid和归一化标题双重去重同款不同链接是选品数据里最大的坑、数值字段统一转 float价格字段在 TOP 返回里是字符串1688 批发价单位是分不是元混用时不换算毛利率能算出负一万。然后是三个核心指标。这三个指标没有标准答案下面是可跑的起点公式系数请拿你自己类目的历史数据校准# metrics.py —— 指标计算竞争度 / 毛利率 / 动销增速 import math, re NOISE re.compile(r(包邮|特价|拍下|备注|新款|爆款|清仓|官方|旗舰店)) def norm_title(t: str) - str: return NOISE.sub(, re.sub(r[^\w\u4e00-\u9fff], , t or )) def competition(same_cnt: int, cat_item_total: int) - float: 竞争度 0~1同款在售数占类目商品量比例log 压缩防头部垄断 if cat_item_total 0: return 1.0 return min(1.0, math.log1p(same_cnt) / math.log1p(max(cat_item_total, 2))) def gross_margin(retail: float, cost: float, fee_rate: float 0.05) - float: 毛利率(零售-拿货-平台扣点)/零售fee_rate 按类目佣金物流损耗估 if retail 0: return -1.0 return (retail - cost - retail * fee_rate) / retail def momentum(sales_30: float, sales_prev30: float) - float: 动销增速环比上限截断防新品 0-100 爆炸 if sales_prev30 0: return 0.5 if sales_30 0 else 0.0 return max(-1.0, min(2.0, sales_30 / sales_prev30 - 1)) def clean_and_dedupe(items: list[dict]) - list[dict]: seen, out set(), [] for it in items: key norm_title(it.get(title, )) if not key or key in seen: continue seen.add(key) out.append(it) return outsame_cnt同款在售数物料搜索接口不直接给实操里用同关键词翻页后归一化标题的出现次数做代理变量——这就是数据层的价值指标可以粗糙但口径必须一致且可解释。再给一张起步期的字段口径表团队里每个人对销量的理解不一样是所有数据争吵的源头指标本系列口径常见歧义价格券后到手价zk_final_price划线价/日常价/活动价销量近 30 天件数接口uv字段需确认口径月销 vs 累计销、件 vs 人成本1688 一件代发价含首重运费批发价不含运费、有 MOQ竞争度归一化标题重复数的 log 压缩值搜索结果数/在售商家数每次全链路跑完把接口原始返回按run_id存一份 JSON 快照。两周后你一定会怀疑这个数当时到底是不是这样快照就是数据层的法庭证据——第 15 篇讲监控时这就是数据可追溯的最简实现。五、决策层硬打分先行LLM 只做软评整个候选池的收敛过程就是下面这个漏斗——每一层收窄的手段不同花钱的地方也不同打分器是纯代码不花钱、可复现、可审计——能用公式解决的绝不烧 token。LLM 放在公式筛出的 Top30 之后做公式做不了的语义判断。这个顺序反过来就是灾难先让大模型点评 200 个商品token 费用翻七倍不说模型还会一本正经地推荐毛利率为负的款——它看不到你的账本只看得到标题。权重(0.4, 0.35, 0.25)的含义是毛利优先、动销其次、竞争度兜底。不同生意模式该调不同权重——铺货型卖家可以把动销提到 0.5 赚快钱品牌型卖家应把竞争度权重调高避开红海。建议把权重做成配置文件而不是硬编码每周复盘时用实际动销数据回测一次这就是最朴素的策略迭代。# score.py —— 加权打分 排名 def hard_score(row: dict, w(0.4, 0.35, 0.25)) - float: 输入为已归一化到 0~1 的 毛利分/动销分/低竞争分输出综合分 return round(w[0] * row[m_norm] w[1] * row[g_norm] w[2] * row[c_norm], 4) def normalize(rows: list[dict]) - list[dict]: for field, out in ((margin, m_norm), (growth, g_norm), (comp, c_norm)): vals [r[field] for r in rows] lo, hi min(vals), max(vals) span (hi - lo) or 1.0 for r in rows: v (r[field] - lo) / span r[out] v if field ! comp else 1 - v # 竞争度越低越好取反 return rowsLLM 软评用百炼的 OpenAI 兼容接口对 Top30 逐条输出卖点一句话 最大风险 供应链难度结构化 JSON 返回方便直接进报告# soft_review.py —— LLM 软评qwen-plus价格以百炼官方页为准 import json from openai import OpenAI client OpenAI(base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, api_keyYOUR_DASHSCOPE_KEY) PROMPT 你是资深电商选品分析师。基于以下商品数据输出一行 JSON不要其他内容 {{title:{title},sell_point:卖点一句话,risk:最大风险,supply:低/中/高}} 商品{title} 券后价{price}元 拿货成本约{cost}元 近30天销量{sales} 竞争度{comp:.2f} def soft_review(top_rows: list[dict]) - list[dict]: out [] for r in top_rows: prompt PROMPT.format(titler[title], pricer[price], costr[cost], salesr[sales_30], compr[comp]) resp client.chat.completions.create( modelqwen-plus, temperature0.2, messages[{role: user, content: prompt}]) try: out.append(json.loads(resp.choices[0].message.content)) except json.JSONDecodeError: out.append({title: r[title], sell_point: , risk: 模型输出解析失败, supply: 中}) return out把这两步接进 LangGraph 图时记得第 6 篇的铁律打分节点保持纯函数可重放LLM 调用放独立节点并带幂等键num_iid 日期断点恢复时不重复计费。六、把确认闸口接进来报告发出前最后一道 human-in-the-loop选品报告是 L1 级动作只读分析但本实战里加了一个真实的 L2 动作入围款确认后自动向 1688 供应商发起询价询盘示例保留接口位不代发消息。这一步必须走第 6 篇的确认机制直接复用verify_and_resume状态机# graph.py —— 选品主图采集-清洗-打分-软评-确认-交付 from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.types import interrupt def approval_gate(state: ProductState): short state[shortlist][:8] decision interrupt({ type: selection_review, items: [{num_iid: r[num_iid], title: r[title], score: r[score], risk: r[risk]} for r in short], next_action: 对确认款发起1688供应商询价, ttl_minutes: 120, }) # decision: {approved: [num_iid...], rejected: [...], operator: uid} return {approved: decision[approved], operator: decision[operator]} def build_graph(): g StateGraph(ProductState) g.add_node(expand, expand_keywords) # L1 意图 g.add_node(collect, collect_all) # L2 采集Throttle 内置 g.add_node(metrics, compute_metrics) # L3 数据 g.add_node(score, rank_top30) # L4 决策-硬 g.add_node(review, soft_review_node) # L4 决策-软 g.add_node(gate, approval_gate) # L4 确认闸口 g.add_node(report, deliver_report) # L5 交付 for a, b in [(expand,collect),(collect,metrics),(metrics,score), (score,review),(review,gate),(gate,report)]: g.add_edge(a, b) g.add_edge(START, expand); g.add_edge(report, END) return g.compile(checkpointerSqliteSaver.from_conn_string(xuanpin.db))确认卡片推到飞书/企微的渲染代码第 5、6 篇已给全这里不重复。注意interrupt的 payload 里带上next_action字段——审批人要知道点确认之后会发生什么这是第 6 篇防无脑点确认的设计在业务里的落点。选品场景的审批还有个特殊处时效性。竞品今天 8 强明天可能就 20 强审批拖三天结果就作废了。所以ttl_minutes设 120 而不是第 6 篇示例里的 1440超时未审批的动作是整批作废 通知发起人绝不是默认通过——任何把超时兜底设成自动放行的设计等于闸口形同虚设。顺带说下expand_keywords意图层的实现它是第 1 篇结构化输出的直接复用——让 qwen-turbo 把宠物智能用品扩展成[自动喂食器,智能饮水机,宠物摄像头,智能猫砂盆]并各带价格带建议输出用 JSON schema 约束死。这一步别过度省钱turbo 档跑一次意图拓展不到一分钱但关键词拓展的质量决定候选池的天花板值得为它多花两厘上一个更好的模型。七、交付层Excel 报告与图文推送最终交付两张给运营看的 Excel 明细、给老板看的图文摘要。# report.py —— Excel 选品报告openpyxl from openpyxl import Workbook from openpyxl.styles import Font, PatternFill HEAD [排名,商品ID,标题,券后价,拿货价,毛利率,竞争度, 近30天销量,综合分,卖点,风险,供应链] def write_report(rows: list[dict], path: str) - str: wb Workbook(); ws wb.active; ws.title 选品入围 ws.append(HEAD) for c in ws[1]: c.font Font(boldTrue, colorFFFFFF) c.fill PatternFill(solid, fgColor4472C4) for i, r in enumerate(rows, 1): ws.append([i, r[num_iid], r[title], r[price], r[cost], round(r[margin], 3), round(r[comp], 3), r[sales_30], r[score], r.get(sell_point,), r.get(risk,), r.get(supply,)]) for col, w in zip(ABCDEFGHIJKL, (5,14,40,9,9,8,8,11,8,30,30,8)): ws.column_dimensions[col].width w ws.freeze_panes A2 wb.save(path) return path图文摘要走第 5 篇的渠道层Markdown 表格 每张入围款主图企微侧用template_cardtext_notice 型拼八条要点推送。报告文件建议同时落一份带日期的归档目录每周复盘时把实际动销结果回填进去——这份数据以后就是你调打分权重的训练集也是第 15 篇监控主题的伏笔。报告的排版服务于一个动线老板 30 秒内要能拍板。所以图文摘要只放三样东西——入围款清单图价格综合分、每款一句话卖点、以及最刺眼的风险标注比如该类目 Top3 已被品牌方投诉侵权。Excel 明细是给运营执行用的放全量字段。两个文件同一封推送发出各取所需。千万别把 12 列大表直接怼进聊天卡片——那是把决策成本转嫁给了审批人第 6 篇说过审批人懒得看就会乱点乱点闸口就失效。八、常见报错、调试清单与成本估算8.1 常见报错与调试清单现象 / 原因 / 解法sign error/isv.isv-sign-error— 签名不匹配 — 检查参数 ASCII 升序、空值是否剔除、timestamp 与服务器误差是否超 15 分钟、secret 是否用了别的环境的用官方 API 调试台比对一次签名串。ISP_FLOW_CONTROL_LIMIT/ 41 受限— 触发 QPS 或日额度 — 个人号日额度本就低上Throttle主动限速、减小 page_size、把多关键词改为错峰跑额度不够去控制台看提额路径。isv.missing-parameter:adzone_id/ PID 不匹配— 联盟推广位未申请或与 AppKey 不同主体 — 回联盟后台确认 PID 三段时间码与应用绑定关系重新生成推广位。返回 0 条但 HTTP 200— 筛选条件过严价格带佣金率天猫互斥条件叠加或material_id过期 — 逐条放宽定位物料搜索的推荐位数据本身有时效。result_list.map_data解析 KeyError— 单条数据时 JSON 结构退化 — 解析处统一_as_list(v)dict 包一层 list。毛利率全是负数— 1688 价格单位是分、零售字段是元或把划线价当券后价 — 打印 3 条原始值核对单位统一在数据层转元。同款商品刷屏 Top 榜— 只按 num_iid 去重没做标题归一 — 用norm_title二次去重仍顽固可加主图感知哈希第 18 篇展开。LLM 软评输出不是合法 JSON— 模型加了 markdown 围栏或解释文字 — temperature 降到 0.2、prompt 里不要其他内容、解析失败走兜底行上文代码已含。断点恢复后重复推送确认卡片— interrupt 节点不幂等 — 卡片外发前查pending_approval表同 thread_id 已有未决审批就复用卡片 ID。interrupt() outside of a graph— 在图外直接调了含 interrupt 的函数 — 单元测试里 mock 掉闸口节点集成测试必须经invoke(config{configurable:{thread_id:...}})。审批超时后 resume 报找不到 checkpoint— SQLite 文件被清理或换了目录 — 检查 SqliteSaver 路径是绝对路径且随部署持久化第 6 篇坑 9 同款。报告数字与接口返回对不上— 各层用了不同时间快照 — 全链路带run_id采集时落原始 JSON 快照报告页脚标注数据抓取时间。8.2 成本估算项目口径月成本个人跑通版淘宝联盟 API免费接口个人日额度内0 元1688 开放平台企业主体资源包0起步用 CSV~2980 元/年qwen-plus 软评Top30×每条约 600 token日 2 轮输入 0.8/输出 4.8 元每百万 token阶梯价以百炼官方页为准约 2~5 元qwen-turbo 意图拓展日 2 轮各千余 token不足 1 元服务器本地跑不需要云端最小实例0~40 元合计个人验证期约 5~10 元/月表中 LLM 与云资源数字为估算口径发布前请实测回填API 价格变动频繁以阿里云百炼官方计费页为准。时间成本上从接口申请到全链路跑通主要卡在开放平台审核1-3 个工作日和权限申请代码本身两天内能闭环——所以第二节说的 CSV 兜底路线不是妥协是把等审核变成边等边开发的正确姿势。总结这一篇把系列前六篇真正串成了线——意图解析是第 1 篇的工具调用图编排是第 2 篇的 LangGraph指标口径是第 3 篇知识库要可信的变体分层调试是第 4 篇的方法论渠道推送是第 5 篇确认闸口是第 6 篇。而它自己贡献的新知识点是业务 Agent 的天花板往往不在模型在数据合规和指标口径。数据源选错全链路白干口径不一致打分再花哨也是自嗨。先把 CSV 版跑通、再换官方接口、最后才谈调权重加模型——这个顺序能让你在审核等待期一直有活干。下一篇预告《自动客服 Agent 实战》——从帮你做决策转向替你接待人意图分流、知识库兜底、转人工策略、会话状态管理以及客服场景独有的确认设计退款动作怎么管。如果这篇实战帮你少走了合规弯路点个赞收藏支持一下系列前六篇在主页合集里从第 1 篇顺读体验最佳。你在选品/数据合规上踩过什么坑评论区聊聊点赞多的坑我会写进后续篇章的报错清单。