1. 电商运营的重复劳动到底卡在哪一步如果你做过电商运营大概率经历过这种节奏大促前一周每天先花两三个小时翻竞品榜单、拉价格带、算毛利再花一两个小时写标题和卖点接着处理主图、填属性、做合规检查最后上架。单看每个环节都不难但串起来就是一条又长又碎的流水线任何一环卡住整批上新就得往后拖。问题的根子不在“有没有 AI 工具”而在“工具之间没有工程化的编排”。市面上写标题的、作图的、算数据的工具很多可它们各自为战选品脚本输出的 CSV 要手动喂给文案工具文案结果又要手动贴进上架后台中间还夹着人工核对违规词。单点提效 30%整体流程却因为数据搬运和人工校验实际提效可能连 10% 都不到。这篇要解决的就是这条链路的“编排层”问题。我会用 AI Agent Harness Engineering 的思路把选品分析、竞品调研、内容生成、素材处理、合规校验、自动上架这几个环节做成一套可复制、可监控、可重试的 Agent 流水线。所谓 Harness Engineering你可以理解成给一群 Agent 配一个“调度中枢”谁先跑、谁依赖谁、失败了怎么重试、成本超了怎么熔断、每一步的日志去哪查全部由这个中枢统一管。适合谁看有 Python 基础、了解大模型 API 调用、正在被重复上新流程拖住的电商运营或中小商家技术负责人。读完之后你能拿到一份可落地的 config.toml 与 settings.json 骨架以及分阶段的验证动作在自己的环境里把“选品到上架”跑通。下面所有模型调用统一走 TaoToken 的兼容接口省去多平台密钥来回切换的麻烦。2. 前置准备TaoToken 接入与 Harness 骨架2.1 为什么用 TaoToken 做统一模型入口Harness 里会有多个 Agent 反复调用大模型选品要算蓝海指数、竞品要提炼卖点、内容要生成标题、合规要判断违规词。如果每个 Agent 各接一家模型、各配一套密钥密钥管理和成本统计会非常乱。TaoToken 提供 OpenAI 兼容的接口形态把模型调用收敛到一个 base_url 和一个 key 上Harness 的成本模块只需要统计这一条出口即可。接入信息如下先记下来后面配置会用到官网入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 基址https://taotoken.net/api密钥管理页https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite注意API 基址不要带 UTM 参数密钥和文档链接按上面写法带上方便后续排查问题时定位来源。2.2 环境与依赖Python 建议 3.10 以上Redis 用于任务队列和幂等去重。依赖装这些就够跑通主链路pip install openai1.40.0 redis5.0.8 pydantic2.8.2 \ pydantic-settings2.4.0 requests2.32.3 \ pillow10.4.0 pandas2.2.2 tenacity8.5.0tenacity用来做重试pydantic-settings负责从环境变量和配置文件加载配置比手写os.getenv干净很多。2.3 config.tomlHarness 的全局参数把调度策略、重试、成本阈值都放进config.tomlAgent 代码里不写死任何魔法数字# config.toml [harness] max_retry 3 retry_backoff_seconds 2 task_timeout_seconds 300 cost_limit_per_task 2.0 # 单任务模型成本上限元 daily_cost_limit 200.0 # 单日总成本上限元 log_level INFO log_file harness.log [queue] redis_host 127.0.0.1 redis_port 6379 redis_db 0 queue_key harness:tasks dead_letter_key harness:dead [llm] base_url https://taotoken.net/api model gpt-4o-mini temperature 0.3 timeout 60 [alert] webhook # 企业微信/钉钉机器人地址留空则不告警2.4 settings.json密钥与平台参数密钥类信息单独放settings.json并加入.gitignore避免误提交{ llm_api_key: sk-你的TaoToken密钥, platform: { app_key: your-platform-app-key, app_secret: your-platform-app-secret, base_url: https://openapi-your-platform.com }, image_api_key: your-image-api-key, oss_bucket: your-bucket }2.5 加载配置并初始化 Harness# harness_boot.py import json, logging, redis from pathlib import Path from pydantic_settings import BaseSettings class Settings(BaseSettings): llm_api_key: str platform: dict image_api_key: str oss_bucket: str classmethod def load(cls, pathsettings.json): return cls(**json.loads(Path(path).read_text(encodingutf-8))) settings Settings.load() logging.basicConfig( levelINFO, format%(asctime)s | %(name)s | %(levelname)s | %(message)s, handlers[logging.FileHandler(harness.log), logging.StreamHandler()], ) logger logging.getLogger(harness) redis_client redis.Redis(host127.0.0.1, port6379, db0, decode_responsesTrue) class Harness: def __init__(self, settings, redis_client): self.settings settings self.redis redis_client self.tools {} self.agents {} self.cost_today 0.0 def register_tool(self, name): def deco(fn): self.tools[name] fn return fn return deco def register_agent(self, name): def deco(cls): self.agents[name] cls(self) return cls return deco def push(self, queue_name, payload): import json as _json self.redis.lpush(fharness:{queue_name}, _json.dumps(payload, ensure_asciiFalse)) harness Harness(settings, redis_client) logger.info(Harness 初始化完成模型出口%s, settings.llm_api_key[:6] ***)到这里骨架就搭好了。接下来把模型调用和平台调用注册成工具Agent 只依赖工具名不直接碰密钥。3. 可复制配置工具注册与 Agent 编排3.1 注册模型调用工具带成本统计与重试from openai import OpenAI from tenacity import retry, stop_after_attempt, wait_exponential client OpenAI(api_keysettings.llm_api_key, base_urlhttps://taotoken.net/api) harness.register_tool(call_llm) retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def call_llm(prompt: str, response_format: str text): resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0.3, response_format{type: json_object} if response_format json else None, ) usage resp.usage.total_tokens cost usage * 0.0015 / 1000 # 按实际单价换算 harness.cost_today cost if harness.cost_today 200.0: raise RuntimeError(单日成本超限任务熔断) content resp.choices[0].message.content return json.loads(content) if response_format json else contentretry处理的是网络抖动和限流Harness 层的max_retry处理的是业务级失败比如模型返回格式不对两层职责分开排障时不会互相干扰。3.2 注册平台 API 工具harness.register_tool(call_platform_api) def call_platform_api(api_path: str, params: dict) - dict: url settings.platform[base_url] api_path headers {X-App-Key: settings.platform[app_key]} resp requests.post(url, jsonparams, headersheaders, timeout30) resp.raise_for_status() return resp.json()3.3 选品分析 Agent选品用蓝海指数做初筛公式是搜索热度 × 平均转化率 × 100 / (在售商品数 × 均价系数)均价系数取min(商品价/品类均价, 1)。Agent 只负责把数据交给模型判断阈值判断留在代码里避免模型“自由发挥”。harness.register_agent(product_selection) class ProductSelectionAgent: def __init__(self, h): self.h h PROMPT 你是电商选品专家。根据以下商品数据筛选蓝海商品。 规则蓝海指数3毛利率30%排除侵权与风险品类。 数据{goods} 输出 JSON 数组字段goods_id, goods_name, price, profit_rate, blue_ocean_index, core_tags。 def run(self, platform: str, category: str, top_n: int 100): hot self.h.tools[call_platform_api]( /goods/hotList, {platform: platform, category: category, top_n: top_n}) safe [g for g in hot.get(list, []) if not g.get(is_risk)] result self.h.tools[call_llm]( self.PROMPT.format(goodsjson.dumps(safe, ensure_asciiFalse)), response_formatjson) for goods in result: self.h.push(competitor_research, goods) return result3.4 竞品调研与内容生成 Agent竞品调研把 Top10 竞品的卖点和差评拼成上下文让模型输出差异化卖点和建议售价内容生成再基于这些结论产出标题、短标题、详情文案和作图提示词。两个 Agent 之间通过队列传递不直接函数调用这样任一环节失败都能单独重放。harness.register_agent(competitor_research) class CompetitorResearchAgent: def __init__(self, h): self.h h def run(self, goods: dict): comps self.h.tools[call_platform_api]( /goods/competitorList, {goods_id: goods[goods_id], top_n: 10}) ctx \n.join( f{c[goods_name]} | 价{c[price]} | 卖点{c.get(sell_point,)} | 评价{c.get(comments,)} for c in comps.get(list, [])) prompt f基于竞品数据输出 JSON {{共性卖点:[], 未满足需求:[], 差异化卖点:[], 建议售价:0, 核心关键词:[]}} 竞品数据{ctx} result self.h.tools[call_llm](prompt, response_formatjson) result[base_goods] goods self.h.push(content_generate, result) return resultharness.register_agent(content_generate) class ContentGenerateAgent: def __init__(self, h): self.h h def run(self, research: dict): prompt f为商品生成上架内容输出 JSON {{title:,short_title:,detail_content:,main_image_text:[],image_prompt:}} 要求标题≤30字含核心关键词无极限词详情分痛点/卖点/保障三段。 关键词{research[核心关键词]} 差异化卖点{research[差异化卖点]} 建议售价{research[建议售价]} content self.h.tools[call_llm](prompt, response_formatjson) content[base_info] research self.h.push(compliance_check, content) return content3.5 合规校验与自动上架 Agent合规校验做两层本地极限词表先扫一遍再让模型判断整体合规性最后调平台校验接口。三层都过才允许上架任何一层失败就进死信队列等人工处理。FORBIDDEN [最, 第一, 顶级, 国家级, 最佳, 绝对] harness.register_agent(compliance_check) class ComplianceCheckAgent: def __init__(self, h): self.h h def run(self, content: dict): text content[title] content[detail_content] hit [w for w in FORBIDDEN if w in text] if hit: raise ValueError(f命中极限词{hit}) verdict self.h.tools[call_llm]( f判断以下内容是否违反广告法或平台规则只回 yes 或 no{text}) if verdict.strip().lower() ! yes: raise ValueError(模型判定不合规) self.h.push(publish_goods, content) return contentharness.register_agent(publish_goods) class PublishGoodsAgent: def __init__(self, h): self.h h def run(self, content: dict): params { goods_name: content[title], short_name: content[short_title], price: content[base_info][建议售价], stock: 1000, main_images: content.get(main_images, []), detail_content: content[detail_content], } resp self.h.tools[call_platform_api](/goods/create, params) if resp.get(code) ! 0: raise RuntimeError(f上架失败{resp.get(msg)}) logger.info(上架成功 goods_id%s, resp[goods_id]) return resp4. 验证请求分阶段跑通并确认结果不要一上来就跑全链路按下面三个阶段验证每阶段确认输出再进下一步。4.1 阶段一验证模型出口连通先用一条最小请求确认 TaoToken 的 base_url 和 key 可用curl https://taotoken.net/api/chat/completions \ -H Authorization: Bearer sk-你的密钥 \ -H Content-Type: application/json \ -d {model:gpt-4o-mini,messages:[{role:user,content:回复 ok}]}返回里能看到choices[0].message.content为ok说明模型出口正常。如果这里报 401去密钥管理页核对 key 是否复制完整。4.2 阶段二单 Agent 验证只跑选品 Agent确认它能拿到榜单数据并返回结构化 JSONfrom harness_boot import harness agent harness.agents[product_selection] result agent.run(platformdouyin, category女装连衣裙, top_n50) print(len(result), result[:1])预期输出是筛选后的商品列表每条带blue_ocean_index和profit_rate。如果返回空列表先看榜单接口是否真的返回了数据再看模型是否把全部商品判为不合格。4.3 阶段三全链路验证把队列消费循环启动观察任务从选品一路流转到上架import json, time CHAIN [product_selection, competitor_research, content_generate, compliance_check, publish_goods] def worker(): while True: for name in CHAIN: raw harness.redis.rpop(fharness:{name}) if not raw: continue payload json.loads(raw) try: harness.agents[name].run(**payload) if name product_selection \ else harness.agents[name].run(payload) except Exception as e: logger.error(任务失败 %s: %s, name, e) harness.redis.lpush(harness:dead, raw) time.sleep(1) if __name__ __main__: harness.push(product_selection, {platform: douyin, category: 女装连衣裙, top_n: 50}) worker()成功时harness.log里会依次出现选品完成、竞品调研完成、内容生成完成、合规通过、上架成功 goods_idxxx 的日志。看到 goods_id 就说明全链路通了。5. 本篇常见错排查5.1 模型返回不是合法 JSON现象json.loads抛JSONDecodeError。原因通常是 prompt 里没强调“只输出 JSON”或者模型在 JSON 前后加了说明文字。解决在 prompt 末尾加“只输出 JSON不要任何解释”并开启response_format{type:json_object}。如果仍偶发在工具层加一次兜底解析截取第一个{到最后一个}之间的内容再解析。5.2 任务在队列里反复重试现象同一个任务在日志里出现多次。原因多半是 Agent 内部抛异常后 Harness 重试但异常本身是数据问题比如商品缺goods_id重试也不会成功。解决区分可重试异常网络、限流和不可重试异常数据缺失、合规失败后者直接进死信队列不要浪费重试次数。5.3 成本统计对不上现象cost_today和实际账单差异大。原因通常是只统计了成功请求重试和失败的请求没算。解决把成本累加放在call_llm的最外层无论成功失败都记一次或者按请求发起次数统计。5.4 上架接口返回签名错误现象call_platform_api返回 401 或签名不匹配。原因不同平台签名算法不同示例里省略了签名实现。解决把签名逻辑封装成独立函数在call_platform_api里统一调用并先用平台提供的调试工具验证签名串。5.5 合规校验误杀正常文案现象正常标题被判违规。原因极限词表过于宽泛比如“最”会命中“最近上新”。解决极限词用正则加边界匹配或者改成“最形容词”的组合模式减少误伤。6. 继续往下走把链路跑稳再跑快链路跑通只是第一步。真正上线前建议先做两件事一是给每个任务加唯一task_id所有环节日志都带上它出问题时能一键串起整条链路二是把死信队列接上告警任务进死信就推消息到群里别等第二天才发现昨晚的上新全卡住了。如果你在接入阶段遇到密钥或 base_url 的问题可以直接去 API Keys 页面重新生成一个 key 对比测试模型返回格式不稳定时用模型对话页面手动发几条 prompt 观察输出规律比在代码里反复试快得多。等链路稳定、要长期跑批量上新和 Agent 常驻调度时再考虑 Coding Plan 这类面向持续编码和 Agent 运行的方案把成本压下来。我自己的经验是先把选品到上架的单条链路跑通并稳定运行一周再考虑加自动定价、自动客服这些扩展环节。链路没跑稳就急着加功能最后排障成本会翻倍。
AI Agent Harness Engineering 在电商运营中的实战：从选品分析到自动上架的全链路自动化
1. 电商运营的重复劳动到底卡在哪一步如果你做过电商运营大概率经历过这种节奏大促前一周每天先花两三个小时翻竞品榜单、拉价格带、算毛利再花一两个小时写标题和卖点接着处理主图、填属性、做合规检查最后上架。单看每个环节都不难但串起来就是一条又长又碎的流水线任何一环卡住整批上新就得往后拖。问题的根子不在“有没有 AI 工具”而在“工具之间没有工程化的编排”。市面上写标题的、作图的、算数据的工具很多可它们各自为战选品脚本输出的 CSV 要手动喂给文案工具文案结果又要手动贴进上架后台中间还夹着人工核对违规词。单点提效 30%整体流程却因为数据搬运和人工校验实际提效可能连 10% 都不到。这篇要解决的就是这条链路的“编排层”问题。我会用 AI Agent Harness Engineering 的思路把选品分析、竞品调研、内容生成、素材处理、合规校验、自动上架这几个环节做成一套可复制、可监控、可重试的 Agent 流水线。所谓 Harness Engineering你可以理解成给一群 Agent 配一个“调度中枢”谁先跑、谁依赖谁、失败了怎么重试、成本超了怎么熔断、每一步的日志去哪查全部由这个中枢统一管。适合谁看有 Python 基础、了解大模型 API 调用、正在被重复上新流程拖住的电商运营或中小商家技术负责人。读完之后你能拿到一份可落地的 config.toml 与 settings.json 骨架以及分阶段的验证动作在自己的环境里把“选品到上架”跑通。下面所有模型调用统一走 TaoToken 的兼容接口省去多平台密钥来回切换的麻烦。2. 前置准备TaoToken 接入与 Harness 骨架2.1 为什么用 TaoToken 做统一模型入口Harness 里会有多个 Agent 反复调用大模型选品要算蓝海指数、竞品要提炼卖点、内容要生成标题、合规要判断违规词。如果每个 Agent 各接一家模型、各配一套密钥密钥管理和成本统计会非常乱。TaoToken 提供 OpenAI 兼容的接口形态把模型调用收敛到一个 base_url 和一个 key 上Harness 的成本模块只需要统计这一条出口即可。接入信息如下先记下来后面配置会用到官网入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 基址https://taotoken.net/api密钥管理页https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite注意API 基址不要带 UTM 参数密钥和文档链接按上面写法带上方便后续排查问题时定位来源。2.2 环境与依赖Python 建议 3.10 以上Redis 用于任务队列和幂等去重。依赖装这些就够跑通主链路pip install openai1.40.0 redis5.0.8 pydantic2.8.2 \ pydantic-settings2.4.0 requests2.32.3 \ pillow10.4.0 pandas2.2.2 tenacity8.5.0tenacity用来做重试pydantic-settings负责从环境变量和配置文件加载配置比手写os.getenv干净很多。2.3 config.tomlHarness 的全局参数把调度策略、重试、成本阈值都放进config.tomlAgent 代码里不写死任何魔法数字# config.toml [harness] max_retry 3 retry_backoff_seconds 2 task_timeout_seconds 300 cost_limit_per_task 2.0 # 单任务模型成本上限元 daily_cost_limit 200.0 # 单日总成本上限元 log_level INFO log_file harness.log [queue] redis_host 127.0.0.1 redis_port 6379 redis_db 0 queue_key harness:tasks dead_letter_key harness:dead [llm] base_url https://taotoken.net/api model gpt-4o-mini temperature 0.3 timeout 60 [alert] webhook # 企业微信/钉钉机器人地址留空则不告警2.4 settings.json密钥与平台参数密钥类信息单独放settings.json并加入.gitignore避免误提交{ llm_api_key: sk-你的TaoToken密钥, platform: { app_key: your-platform-app-key, app_secret: your-platform-app-secret, base_url: https://openapi-your-platform.com }, image_api_key: your-image-api-key, oss_bucket: your-bucket }2.5 加载配置并初始化 Harness# harness_boot.py import json, logging, redis from pathlib import Path from pydantic_settings import BaseSettings class Settings(BaseSettings): llm_api_key: str platform: dict image_api_key: str oss_bucket: str classmethod def load(cls, pathsettings.json): return cls(**json.loads(Path(path).read_text(encodingutf-8))) settings Settings.load() logging.basicConfig( levelINFO, format%(asctime)s | %(name)s | %(levelname)s | %(message)s, handlers[logging.FileHandler(harness.log), logging.StreamHandler()], ) logger logging.getLogger(harness) redis_client redis.Redis(host127.0.0.1, port6379, db0, decode_responsesTrue) class Harness: def __init__(self, settings, redis_client): self.settings settings self.redis redis_client self.tools {} self.agents {} self.cost_today 0.0 def register_tool(self, name): def deco(fn): self.tools[name] fn return fn return deco def register_agent(self, name): def deco(cls): self.agents[name] cls(self) return cls return deco def push(self, queue_name, payload): import json as _json self.redis.lpush(fharness:{queue_name}, _json.dumps(payload, ensure_asciiFalse)) harness Harness(settings, redis_client) logger.info(Harness 初始化完成模型出口%s, settings.llm_api_key[:6] ***)到这里骨架就搭好了。接下来把模型调用和平台调用注册成工具Agent 只依赖工具名不直接碰密钥。3. 可复制配置工具注册与 Agent 编排3.1 注册模型调用工具带成本统计与重试from openai import OpenAI from tenacity import retry, stop_after_attempt, wait_exponential client OpenAI(api_keysettings.llm_api_key, base_urlhttps://taotoken.net/api) harness.register_tool(call_llm) retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def call_llm(prompt: str, response_format: str text): resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0.3, response_format{type: json_object} if response_format json else None, ) usage resp.usage.total_tokens cost usage * 0.0015 / 1000 # 按实际单价换算 harness.cost_today cost if harness.cost_today 200.0: raise RuntimeError(单日成本超限任务熔断) content resp.choices[0].message.content return json.loads(content) if response_format json else contentretry处理的是网络抖动和限流Harness 层的max_retry处理的是业务级失败比如模型返回格式不对两层职责分开排障时不会互相干扰。3.2 注册平台 API 工具harness.register_tool(call_platform_api) def call_platform_api(api_path: str, params: dict) - dict: url settings.platform[base_url] api_path headers {X-App-Key: settings.platform[app_key]} resp requests.post(url, jsonparams, headersheaders, timeout30) resp.raise_for_status() return resp.json()3.3 选品分析 Agent选品用蓝海指数做初筛公式是搜索热度 × 平均转化率 × 100 / (在售商品数 × 均价系数)均价系数取min(商品价/品类均价, 1)。Agent 只负责把数据交给模型判断阈值判断留在代码里避免模型“自由发挥”。harness.register_agent(product_selection) class ProductSelectionAgent: def __init__(self, h): self.h h PROMPT 你是电商选品专家。根据以下商品数据筛选蓝海商品。 规则蓝海指数3毛利率30%排除侵权与风险品类。 数据{goods} 输出 JSON 数组字段goods_id, goods_name, price, profit_rate, blue_ocean_index, core_tags。 def run(self, platform: str, category: str, top_n: int 100): hot self.h.tools[call_platform_api]( /goods/hotList, {platform: platform, category: category, top_n: top_n}) safe [g for g in hot.get(list, []) if not g.get(is_risk)] result self.h.tools[call_llm]( self.PROMPT.format(goodsjson.dumps(safe, ensure_asciiFalse)), response_formatjson) for goods in result: self.h.push(competitor_research, goods) return result3.4 竞品调研与内容生成 Agent竞品调研把 Top10 竞品的卖点和差评拼成上下文让模型输出差异化卖点和建议售价内容生成再基于这些结论产出标题、短标题、详情文案和作图提示词。两个 Agent 之间通过队列传递不直接函数调用这样任一环节失败都能单独重放。harness.register_agent(competitor_research) class CompetitorResearchAgent: def __init__(self, h): self.h h def run(self, goods: dict): comps self.h.tools[call_platform_api]( /goods/competitorList, {goods_id: goods[goods_id], top_n: 10}) ctx \n.join( f{c[goods_name]} | 价{c[price]} | 卖点{c.get(sell_point,)} | 评价{c.get(comments,)} for c in comps.get(list, [])) prompt f基于竞品数据输出 JSON {{共性卖点:[], 未满足需求:[], 差异化卖点:[], 建议售价:0, 核心关键词:[]}} 竞品数据{ctx} result self.h.tools[call_llm](prompt, response_formatjson) result[base_goods] goods self.h.push(content_generate, result) return resultharness.register_agent(content_generate) class ContentGenerateAgent: def __init__(self, h): self.h h def run(self, research: dict): prompt f为商品生成上架内容输出 JSON {{title:,short_title:,detail_content:,main_image_text:[],image_prompt:}} 要求标题≤30字含核心关键词无极限词详情分痛点/卖点/保障三段。 关键词{research[核心关键词]} 差异化卖点{research[差异化卖点]} 建议售价{research[建议售价]} content self.h.tools[call_llm](prompt, response_formatjson) content[base_info] research self.h.push(compliance_check, content) return content3.5 合规校验与自动上架 Agent合规校验做两层本地极限词表先扫一遍再让模型判断整体合规性最后调平台校验接口。三层都过才允许上架任何一层失败就进死信队列等人工处理。FORBIDDEN [最, 第一, 顶级, 国家级, 最佳, 绝对] harness.register_agent(compliance_check) class ComplianceCheckAgent: def __init__(self, h): self.h h def run(self, content: dict): text content[title] content[detail_content] hit [w for w in FORBIDDEN if w in text] if hit: raise ValueError(f命中极限词{hit}) verdict self.h.tools[call_llm]( f判断以下内容是否违反广告法或平台规则只回 yes 或 no{text}) if verdict.strip().lower() ! yes: raise ValueError(模型判定不合规) self.h.push(publish_goods, content) return contentharness.register_agent(publish_goods) class PublishGoodsAgent: def __init__(self, h): self.h h def run(self, content: dict): params { goods_name: content[title], short_name: content[short_title], price: content[base_info][建议售价], stock: 1000, main_images: content.get(main_images, []), detail_content: content[detail_content], } resp self.h.tools[call_platform_api](/goods/create, params) if resp.get(code) ! 0: raise RuntimeError(f上架失败{resp.get(msg)}) logger.info(上架成功 goods_id%s, resp[goods_id]) return resp4. 验证请求分阶段跑通并确认结果不要一上来就跑全链路按下面三个阶段验证每阶段确认输出再进下一步。4.1 阶段一验证模型出口连通先用一条最小请求确认 TaoToken 的 base_url 和 key 可用curl https://taotoken.net/api/chat/completions \ -H Authorization: Bearer sk-你的密钥 \ -H Content-Type: application/json \ -d {model:gpt-4o-mini,messages:[{role:user,content:回复 ok}]}返回里能看到choices[0].message.content为ok说明模型出口正常。如果这里报 401去密钥管理页核对 key 是否复制完整。4.2 阶段二单 Agent 验证只跑选品 Agent确认它能拿到榜单数据并返回结构化 JSONfrom harness_boot import harness agent harness.agents[product_selection] result agent.run(platformdouyin, category女装连衣裙, top_n50) print(len(result), result[:1])预期输出是筛选后的商品列表每条带blue_ocean_index和profit_rate。如果返回空列表先看榜单接口是否真的返回了数据再看模型是否把全部商品判为不合格。4.3 阶段三全链路验证把队列消费循环启动观察任务从选品一路流转到上架import json, time CHAIN [product_selection, competitor_research, content_generate, compliance_check, publish_goods] def worker(): while True: for name in CHAIN: raw harness.redis.rpop(fharness:{name}) if not raw: continue payload json.loads(raw) try: harness.agents[name].run(**payload) if name product_selection \ else harness.agents[name].run(payload) except Exception as e: logger.error(任务失败 %s: %s, name, e) harness.redis.lpush(harness:dead, raw) time.sleep(1) if __name__ __main__: harness.push(product_selection, {platform: douyin, category: 女装连衣裙, top_n: 50}) worker()成功时harness.log里会依次出现选品完成、竞品调研完成、内容生成完成、合规通过、上架成功 goods_idxxx 的日志。看到 goods_id 就说明全链路通了。5. 本篇常见错排查5.1 模型返回不是合法 JSON现象json.loads抛JSONDecodeError。原因通常是 prompt 里没强调“只输出 JSON”或者模型在 JSON 前后加了说明文字。解决在 prompt 末尾加“只输出 JSON不要任何解释”并开启response_format{type:json_object}。如果仍偶发在工具层加一次兜底解析截取第一个{到最后一个}之间的内容再解析。5.2 任务在队列里反复重试现象同一个任务在日志里出现多次。原因多半是 Agent 内部抛异常后 Harness 重试但异常本身是数据问题比如商品缺goods_id重试也不会成功。解决区分可重试异常网络、限流和不可重试异常数据缺失、合规失败后者直接进死信队列不要浪费重试次数。5.3 成本统计对不上现象cost_today和实际账单差异大。原因通常是只统计了成功请求重试和失败的请求没算。解决把成本累加放在call_llm的最外层无论成功失败都记一次或者按请求发起次数统计。5.4 上架接口返回签名错误现象call_platform_api返回 401 或签名不匹配。原因不同平台签名算法不同示例里省略了签名实现。解决把签名逻辑封装成独立函数在call_platform_api里统一调用并先用平台提供的调试工具验证签名串。5.5 合规校验误杀正常文案现象正常标题被判违规。原因极限词表过于宽泛比如“最”会命中“最近上新”。解决极限词用正则加边界匹配或者改成“最形容词”的组合模式减少误伤。6. 继续往下走把链路跑稳再跑快链路跑通只是第一步。真正上线前建议先做两件事一是给每个任务加唯一task_id所有环节日志都带上它出问题时能一键串起整条链路二是把死信队列接上告警任务进死信就推消息到群里别等第二天才发现昨晚的上新全卡住了。如果你在接入阶段遇到密钥或 base_url 的问题可以直接去 API Keys 页面重新生成一个 key 对比测试模型返回格式不稳定时用模型对话页面手动发几条 prompt 观察输出规律比在代码里反复试快得多。等链路稳定、要长期跑批量上新和 Agent 常驻调度时再考虑 Coding Plan 这类面向持续编码和 Agent 运行的方案把成本压下来。我自己的经验是先把选品到上架的单条链路跑通并稳定运行一周再考虑加自动定价、自动客服这些扩展环节。链路没跑稳就急着加功能最后排障成本会翻倍。
