这次我们来看一个偏底层但非常实际的问题AI 时代的产品发现凭什么让不同的模型、平台、商家和用户互相协作答案不是再做一个“智能推荐系统”而是先把数据交换和接口调用方式统一成一份开放协议。标题里说的 “An open protocol for AI-mediated product discovery”意思就是为“AI 参与商品发现过程”设计一套公开、可扩展、可审计的交互契约。这篇文章不会去蹭“AI带货”“AI一键成片”这些热词而是从工程角度拆解AI 做产品发现时需要哪些数据、哪些接口、哪些权限控制商家、平台、AI Agent 和普通用户之间怎么通过协议协作如果你要自己实现一个兼容层代码怎么写、接口怎么测、批量任务怎么跑。文章适用这几类读者正在做 AI 电商搜索、AI 导购、Agent 工具链的开发者需要把商品库开放给第三方 AI 应用的产品经理和架构师以及对“AI 如何和现有商业系统对接”感兴趣的技术人员。1. 核心能力速览先给一张速览表明确这款“协议”能做什么、不能做什么。注意这里讨论的是一套通用设计模板不是某个已上线的商业标准。能力项说明协议类型AI Agent 与商品数据源之间的开放接口规范核心能力产品语义搜索、属性过滤、推荐结果返回、批量发现、溯源审计交互方式HTTP JSON API / 事件订阅数据格式JSON Schema支持扩展字段是否支持批量任务支持通过批量请求或任务列表接口是否支持 API支持标准 RESTful API是否支持权限控制支持ApiKey 作用域 scope是否可以本地部署可以作为轻量服务运行显存要求不涉及纯服务端逻辑适合场景AI 导购、商品问答、Agent 工具、跨平台比价、智能荐货不适合场景实时大流量高并发电商主站需要做性能优化从材料看该主题更偏向协议设计和技术方案不涉及具体模型权重或显卡资源。因此文章重点放在接口设计、数据建模、部署验证和批量任务场景上。2. 适用场景与使用边界2.1 适合谁做 AI 电商导购、智能客服、Agent 搜索工具的团队。手握商品库但想开放给第三方 AI 应用调用的平台方。想在多平台间做商品聚合、比价、比较的开发者。研究 AI Agent 工具编排、MCP 类似协议的技术人员。2.2 能解决什么问题现在很多 AI 产品发现能力都是“烟囱式”的每个平台一套商品 API每个模型一套 prompt商品字段五花八门价格单位不一致图片尺寸不一样库存状态表达也不同。开放协议的价值在于统一商品元数据格式让 AI 模型不用适配 N 套数据源。统一搜索和推荐接口让 Agent 用相同方式完成“找商品、比参数、筛价格、看库存、取详情”流程。支持审计与溯源每次推荐结果都能回溯数据来源和调用链。支持批量任务让 AI 在离线场景里完成大规模选品或商品去重。2.3 不适合什么不适合实时秒杀、超低延迟的电商核心交易链路。不适合已有成熟开放平台的内部替换除非你想做跨平台聚合层。不适合把商品数据直接公开协议本身不解决数据脱敏和版权问题。2.4 合规边界使用这套协议时必须注意商品库的版权归属图片、文案、价格数据来源需要有权使用。个人隐私不允许通过商品推荐接口收集用户身份信息。平台规则跨平台抓取或比价需要遵守各平台的服务条款。AI 生成推荐结果需要标识 AI 参与并对推荐偏差负责。商用前进行合规审查特别涉及品牌、价格、肖像和用户数据。3. 环境准备与前置条件虽然这是一个协议设计但落地时需要一个最小可运行实现。下面给出一套通用环境准备清单不锁定具体版本按实际项目替换即可。3.1 操作系统与运行时Linux / macOS / Windows 都可以。推荐 Python 3.10 或更高版本便于使用 Pydantic 和 FastAPI。Node.js 18 也可以如果要实现 Node 版本协议服务。没有 GPU 要求纯 CPU 服务即可。3.2 依赖库建议后端服务pip install fastapi uvicorn pydantic requests客户端调用pip install requests如果要用 OpenAPI 生成文档FastAPI 自带 Swagger UI不需要额外插件。3.3 项目目录结构建议按下面的结构维护代码和配置open-product-discovery/ ├── server.py # 协议服务端 ├── client.py # 客户端调用示例 ├── schemas.py # 商品和请求响应的数据模型 ├── config.yaml # 服务配置端口、限流、超时 ├── sample_products.json # 测试商品数据 └── tests/ ├── test_search.py ├── test_recommend.py └── test_batch.py这套结构不是强制的但它能让你在后续加功能时不混乱。4. 协议设计与数据格式开放协议的核心是数据格式。如果商品字段不统一AI 就没法稳定理解。下面给出一套示例协议设计命名为 “Open Product Discovery Protocol”简称 OPDP用于演示。4.1 商品对象模型一个最小可用的商品对象应该包含以下字段{ product_id: sku_001, title: 机械键盘 87 键 热插拔, description: 支持热插拔轴体RGB 背光适用于办公和游戏, brand: 示例品牌, categories: [电脑外设, 键盘], attributes: { layout: 87, switch_type: mechanical, connection: usb, color: black }, price: { amount: 299, currency: CNY }, stock_status: in_stock, images: [ { url: https://example.com/img/001.jpg, type: main } ], source_url: https://example.com/products/001, updated_at: 2025-06-01T12:00:00Z }这些字段基本能覆盖搜索、过滤、展示和比较的场景。AI Agent 可以不依赖人类可读页面只凭结构化数据完成判断。4.2 搜索请求格式AI Agent 发起的搜索请求需要包含 query、过滤条件、排序方式、分页信息和返回字段。{ query: 适合程序员用的机械键盘, filters: { brand: [示例品牌], price_min: 100, price_max: 500, currency: CNY, stock_status: in_stock }, sort: { field: relevance, order: desc }, page: 1, page_size: 20, fields: [product_id, title, price, attributes, source_url] }协议里要允许 AI Agent 传自然语言 query也允许传结构化 filters。服务端负责把自然语言 query 转成可执行的检索逻辑。4.3 推荐请求格式推荐和搜索不同它基于用户上下文或种子商品做扩展。{ seed_product_ids: [sku_001], context: { purpose: office, budget: 400 }, strategy: similar, limit: 10 }这里strategy可以是similar、complementary、bestseller等具体由服务端实现。4.4 响应格式统一响应结构方便 AI 解析{ request_id: req_20250701_001, items: [ { product_id: sku_001, score: 0.91, product: { product_id: sku_001, title: 机械键盘 87 键 热插拔, price: { amount: 299, currency: CNY } } } ], pagination: { page: 1, page_size: 20, total: 156 }, trace: { strategy: semantic_search, model: builtin_matcher, latency_ms: 45 } }统一响应结构对 AI 特别重要。Agent 不需要“猜”正确字段路径直接按items[i].product.title读取即可。4.5 批量任务协议批量发现场景下单个请求可能遍历大量商品。建议提供任务式接口而不是同步返回大 JSON。请求{ tasks: [ { task_id: batch_001, queries: [办公键盘, 游戏鼠标], filters: { price_max: 500 } } ] }响应{ batch_id: batch_001, status: accepted, check_url: /v1/batches/batch_001 }客户端通过check_url轮询任务状态拿到结果后再做后续聚合。5. 服务端实现与启动方式下面用一个 FastAPI 示例实现一个简化版协议服务端。这里的目的不是生产代码而是给出启动和测试的最小路径。5.1 数据模型schemas.pyfrom pydantic import BaseModel, Field from typing import Optional from typing import List class Price(BaseModel): amount: float currency: str CNY class Product(BaseModel): product_id: str title: str description: Optional[str] brand: Optional[str] categories: List[str] [] attributes: dict {} price: Price stock_status: str in_stock source_url: Optional[str] class SearchFilter(BaseModel): brand: Optional[List[str]] None price_min: Optional[float] None price_max: Optional[float] None currency: Optional[str] CNY stock_status: Optional[str] None class SearchRequest(BaseModel): query: str Field(..., description自然语言搜索词) filters: Optional[SearchFilter] None page: int 1 page_size: int 20 class SearchResult(BaseModel): product_id: str score: float product: Product class SearchResponse(BaseModel): request_id: str items: List[SearchResult] total: int 05.2 服务端server.py省略真实搜索逻辑用一个简单的关键词匹配 过滤示例。import json import time import uuid from fastapi import FastAPI, HTTPException from schemas import Product, SearchRequest, SearchResponse, SearchResult app FastAPI(titleOpen Product Discovery Protocol, version0.1) # 加载示例商品数据 with open(sample_products.json, r, encodingutf-8) as f: PRODUCTS [Product(**item) for item in json.load(f)] app.get(/health) def health(): return {status: ok} app.post(/v1/search, response_modelSearchResponse) def search(req: SearchRequest): start time.time() # 简易过滤 results [] for p in PRODUCTS: if req.filters: if req.filters.price_min is not None and p.price.amount req.filters.price_min: continue if req.filters.price_max is not None and p.price.amount req.filters.price_max: continue if req.filters.brand and p.brand not in req.filters.brand: continue if req.query and req.query.lower() not in p.title.lower() and req.query.lower() not in p.description.lower(): continue results.append(p) # 分页 start_idx (req.page - 1) * req.page_size page_items results[start_idx:start_idx req.page_size] items [SearchResult(product_idp.product_id, score1.0, productp) for p in page_items] return SearchResponse( request_idfreq_{uuid.uuid4().hex[:12]}, itemsitems, totallen(results) ) if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)5.3 示例商品数据sample_products.json[ { product_id: sku_001, title: 机械键盘 87 键 热插拔, description: 支持热插拔轴体RGB 背光, brand: 示例品牌, categories: [电脑外设, 键盘], attributes: {layout: 87}, price: {amount: 299, currency: CNY}, stock_status: in_stock, source_url: https://example.com/products/001 }, { product_id: sku_002, title: 无线鼠标 静音款, description: 适合办公2.4G 连接, brand: 示例品牌, categories: [电脑外设, 鼠标], attributes: {color: gray}, price: {amount: 99, currency: CNY}, stock_status: in_stock, source_url: https://example.com/products/002 } ]5.4 启动服务python server.py启动后访问健康检查http://127.0.0.1:8000/health接口文档http://127.0.0.1:8000/docs搜索接口http://127.0.0.1:8000/v1/searchFastAPI 自带的 Swagger 文档可以直接做手动测试不需要额外写前端页面。6. 客户端调用与批量任务服务端跑起来后下一步就是让 AI Agent 能调用。下面给出一套通用 Python 客户端示例。6.1 搜索调用import requests url http://127.0.0.1:8000/v1/search payload { query: 键盘, filters: { price_max: 400, stock_status: in_stock }, page: 1, page_size: 10 } resp requests.post(url, jsonpayload, timeout10) data resp.json() print(request_id:, data[request_id]) print(total:, data[total]) for item in data[items]: print(item[product][product_id], item[product][title], item[product][price][amount])预期输出request_id: req_xxxx total: 1 sku_001 机械键盘 87 键 热插拔 299.06.2 批量任务轮询假设服务端实现了/v1/batches接口客户端可以这样设计import time import requests BASE http://127.0.0.1:8000 # 提交批量任务 batch_payload { tasks: [ {task_id: batch_001, queries: [键盘, 鼠标], filters: {price_max: 500}} ] } resp requests.post(f{BASE}/v1/batches, jsonbatch_payload, timeout10) batch resp.json() batch_id batch[batch_id] check_url batch[check_url] # 轮询状态 for _ in range(30): status_resp requests.get(f{BASE}{check_url}, timeout10).json() if status_resp[status] completed: for task in status_resp[results]: print(task[task_id], len(task[items]), items) break time.sleep(2) else: print(timeout: 任务未完成)批量任务的核心是“提交 - 轮询 - 拿到结果”。不要让 Agent 长时间挂在一个 HTTP 请求上等结果。6.3 批量任务设计建议每个任务使用独立 task_id方便日志追踪。任务结果落盘到对象存储或数据库而不是保存在内存。设置最大重试次数避免失败任务无限重试。批量任务要有去重逻辑防止重复推荐同一商品。如果处理超时服务端应该返回202 Accepted而不是阻塞在200。7. 功能测试与效果验证拿到一个协议服务验证时不要只看“能返回数据”要按下面的测试维度逐项检查。7.1 健康检查与文档测试目的确认服务正常运行。操作curl http://127.0.0.1:8000/health预期结果{status:ok}7.2 基础搜索测试测试目的验证自然语言搜索和过滤条件是否生效。请求curl -X POST http://127.0.0.1:8000/v1/search \ -H Content-Type: application/json \ -d {query:键盘,filters:{price_max:400}}判断标准返回total大于 0。每个items中的商品标题包含“键盘”或描述相关。商品价格不超过 400。7.3 空结果测试测试目的检查空结果时响应是否正常。请求curl -X POST http://127.0.0.1:8000/v1/search \ -H Content-Type: application/json \ -d {query:不存在的商品名称}预期结果返回total0items[]HTTP 状态码仍然是 200。不能返回 500。7.4 分页测试测试目的验证分页稳定。第 1 页page_size1和第 2 页page_size1不能有重复商品。页码超出范围时返回空列表而不是崩溃。7.5 批量任务测试测试目的验证异步任务流程。提交包含多个 query 的批量任务。轮询check_url确认任务状态最终为completed。检查每个 task 的结果数量和内容是否符合预期。7.6 失败场景注入测试目的确认服务在异常数据下表现稳定。发送price_max为字符串而不是数字。发送缺失query字段的请求。发送超大page_size例如 99999。良好设计下服务端应该返回 422 参数校验错误而不是堆栈异常。7.7 常见失败原因测试现象可能原因排查方式返回 404路由路径错误检查服务端路由注册返回 422请求体字段类型或缺失查看/docs里的 schema返回 500服务端未捕获异常查看 uvicorn 控制台日志返回空结果数据未加载或过滤条件过严检查sample_products.json和搜索逻辑批量任务一直pending异步执行器没跑检查是否启动了后台 worker8. 资源占用与性能观察这个协议服务不涉及 GPU 推理资源占用集中在 CPU 和内存。但如果是真实生产环境仍需要关注性能。8.1 观察指标请求平均延迟建议看 p95而不是平均值。内存占用加载的商品对象数量决定内存量。CPU 使用率搜索逻辑越复杂CPU 占用越高。端口状态确认 8000 端口没有被其他进程占用。8.2 如何测试用简单的并发请求脚本观察import time import requests from concurrent.futures import ThreadPoolExecutor url http://127.0.0.1:8000/v1/search payload {query: 键盘, page_size: 5} def call(_): start time.time() r requests.post(url, jsonpayload, timeout10) return time.time() - start with ThreadPoolExecutor(max_workers20) as ex: latencies list(ex.map(call, range(100))) print(avg latency:, sum(latencies) / len(latencies)) print(max latency:, max(latencies))注意不要用高并发去压本地小服务可能会触发端口耗尽或进程崩溃。8.3 性能优化方向商品数据量大时使用 Redis 或内存索引缓存。搜索接口加参数校验和限流防止被刷。将自然语言 query 转语义向量的部分独立成模型服务不要阻塞主接口。批量任务用消息队列消化而不是在 HTTP worker 里同步执行。8.4 显存思考这个主题不涉及显存。如果你的协议服务内部集成了向量模型做语义搜索那么模型推理部分会有显存需求。但协议层本身没有任何显存占用。建议在架构上把“语义模型推理”和“协议服务”拆开部署协议层保持轻量。9. 常见问题与排查方法下面汇总基于这套通用协议实现中最容易遇到的问题以及处理思路。问题现象可能原因排查方式解决方案依赖安装失败Python 版本不匹配或缺少构建工具检查pip install日志升级 Python 或使用虚拟环境启动后接口文档打不开端口被占用lsof -i :8000或netstat -ano更换--port 8001搜索返回乱序没有定义相关性排序检查返回score值实现排序逻辑或默认按相关度商品数据重复商品 ID 不唯一检查 JSON 数据的product_id导入前做去重批量任务卡在 pending后台 worker 未启动查看任务队列消费者日志启动 worker 增加日志API 调用超时搜索逻辑太慢打印 SQL 或搜索耗时加索引或缓存接口返回 500数据中包含残缺字段检查 pydantic 校验报错修正数据或增加默认值AI Agent 解析失败响应格式不稳定检查request_id和items是否恒定固定 JSON 结构避免字段值类型变化10. 最佳实践与使用建议10.1 先小批量验证协议语义第一次接入时不要直接对接全量商品库。准备 100 个左右覆盖不同类的测试商品先把协议跑通再逐步接入真实数据。这样能快速发现字段映射和查询逻辑的问题。10.2 商品数据要有明确来源与版本协议里的每个商品对象最好带上source_url和updated_at否则 AI Agent 拿到结果后无法溯源。对于比价、推荐场景数据来源和更新时间直接影响结果可信度。10.3 为不同调用方设置不同权限开放协议不等于完全公开。建议至少设置三种角色只读搜索可以查商品但不能看成本价格。批量任务可以发起批量查询但受限频限制。管理端可以维护商品数据。权限通过 ApiKey 和 scope 字段控制代码里校验API_KEYS { test_key: {scopes: [search, batch]} } def check_scope(key, required_scope): if key not in API_KEYS: return False return required_scope in API_KEYS[key][scopes]10.4 日志里不要记录用户隐私协议接口本身不要求传用户身份。如果上游把用户 ID 放在 context 参数里日志输出时要脱敏。不要打印完整用户 id、手机号、地址等字段。10.5 为 AI Agent 设计容错策略AI Agent 调用接口时可能遇到 429 限流、500 错误、网络超时。客户端要做好指数退避重试并缓存上一次成功结果。避免请求失败时直接返回空推荐给用户。def call_with_retry(url, payload, retries3): for i in range(retries): try: resp requests.post(url, jsonpayload, timeout5) resp.raise_for_status() return resp.json() except Exception: time.sleep(2 ** i) return None10.6 版权与合规必须在协议层声明如果你是商品数据提供方建议在协议响应中加入license字段标明数据使用范围。如果协议用于跨境或跨平台商品比较需要先确认当地法律和各平台条款。涉及品牌商标、博主推荐、用户生成内容时必须获得授权。10.7 发布前做一轮 AI 结果抽查AI 产品发现经常会放大错误一个商品标题写错就可能被模型推荐给大量用户。上线前用固定 query 列表跑一遍回归测试记录每次推荐的商品 ID 和分值。发现异常数据及时下架不要指望模型自动纠错。11. 总结与下一步开放协议的核心价值是“让 AI 不再为每个平台手写适配器”。通过统一商品模型、搜索接口、推荐响应和批量任务结构AI Agent 可以把精力放在理解用户意图和优化推荐策略上而不是纠结字段名和接口差异。如果你准备在自己的项目里落地这套思路建议先做三件事定义一份最小商品 JSON Schema只包含你业务里真正用到的字段。写一个只有搜索和详情两个端点的 MVP 服务用 FastAPI 跑通。用一个模拟 Agent 客户端连续调用 50 次记录失败率和返回字段稳定性。最容易踩的坑也提前说清楚一是商品数据字段随意变化导致 Agent 解析报错二是批量任务同步化导致请求超时三是不设权限把商品库直接暴露给公网。后续如果要扩展可以往这几个方向走引入向量检索做语义搜索为每个商品生成 embedding增加订阅推送让 Agent 感知库存和价格变化支持多语言商品描述把协议层做成插件接入现有的 MCPModel Context Protocol或第三方 Agent 框架。协议本身不是银弹但它能把 AI 产品发现的协作边界定清楚。先从一个搜索接口开始把一个字段一个请求打磨稳定比一开始就铺大而全的开放平台更靠谱。建议收藏这篇文章部署时对照检查。 
长河君 · 2026/8/28 12:40:05 · 阅读 1.2k AI产品发现开放协议：统一搜索推荐接口的工程实现
这次我们来看一个偏底层但非常实际的问题AI 时代的产品发现凭什么让不同的模型、平台、商家和用户互相协作答案不是再做一个“智能推荐系统”而是先把数据交换和接口调用方式统一成一份开放协议。标题里说的 “An open protocol for AI-mediated product discovery”意思就是为“AI 参与商品发现过程”设计一套公开、可扩展、可审计的交互契约。这篇文章不会去蹭“AI带货”“AI一键成片”这些热词而是从工程角度拆解AI 做产品发现时需要哪些数据、哪些接口、哪些权限控制商家、平台、AI Agent 和普通用户之间怎么通过协议协作如果你要自己实现一个兼容层代码怎么写、接口怎么测、批量任务怎么跑。文章适用这几类读者正在做 AI 电商搜索、AI 导购、Agent 工具链的开发者需要把商品库开放给第三方 AI 应用的产品经理和架构师以及对“AI 如何和现有商业系统对接”感兴趣的技术人员。1. 核心能力速览先给一张速览表明确这款“协议”能做什么、不能做什么。注意这里讨论的是一套通用设计模板不是某个已上线的商业标准。能力项说明协议类型AI Agent 与商品数据源之间的开放接口规范核心能力产品语义搜索、属性过滤、推荐结果返回、批量发现、溯源审计交互方式HTTP JSON API / 事件订阅数据格式JSON Schema支持扩展字段是否支持批量任务支持通过批量请求或任务列表接口是否支持 API支持标准 RESTful API是否支持权限控制支持ApiKey 作用域 scope是否可以本地部署可以作为轻量服务运行显存要求不涉及纯服务端逻辑适合场景AI 导购、商品问答、Agent 工具、跨平台比价、智能荐货不适合场景实时大流量高并发电商主站需要做性能优化从材料看该主题更偏向协议设计和技术方案不涉及具体模型权重或显卡资源。因此文章重点放在接口设计、数据建模、部署验证和批量任务场景上。2. 适用场景与使用边界2.1 适合谁做 AI 电商导购、智能客服、Agent 搜索工具的团队。手握商品库但想开放给第三方 AI 应用调用的平台方。想在多平台间做商品聚合、比价、比较的开发者。研究 AI Agent 工具编排、MCP 类似协议的技术人员。2.2 能解决什么问题现在很多 AI 产品发现能力都是“烟囱式”的每个平台一套商品 API每个模型一套 prompt商品字段五花八门价格单位不一致图片尺寸不一样库存状态表达也不同。开放协议的价值在于统一商品元数据格式让 AI 模型不用适配 N 套数据源。统一搜索和推荐接口让 Agent 用相同方式完成“找商品、比参数、筛价格、看库存、取详情”流程。支持审计与溯源每次推荐结果都能回溯数据来源和调用链。支持批量任务让 AI 在离线场景里完成大规模选品或商品去重。2.3 不适合什么不适合实时秒杀、超低延迟的电商核心交易链路。不适合已有成熟开放平台的内部替换除非你想做跨平台聚合层。不适合把商品数据直接公开协议本身不解决数据脱敏和版权问题。2.4 合规边界使用这套协议时必须注意商品库的版权归属图片、文案、价格数据来源需要有权使用。个人隐私不允许通过商品推荐接口收集用户身份信息。平台规则跨平台抓取或比价需要遵守各平台的服务条款。AI 生成推荐结果需要标识 AI 参与并对推荐偏差负责。商用前进行合规审查特别涉及品牌、价格、肖像和用户数据。3. 环境准备与前置条件虽然这是一个协议设计但落地时需要一个最小可运行实现。下面给出一套通用环境准备清单不锁定具体版本按实际项目替换即可。3.1 操作系统与运行时Linux / macOS / Windows 都可以。推荐 Python 3.10 或更高版本便于使用 Pydantic 和 FastAPI。Node.js 18 也可以如果要实现 Node 版本协议服务。没有 GPU 要求纯 CPU 服务即可。3.2 依赖库建议后端服务pip install fastapi uvicorn pydantic requests客户端调用pip install requests如果要用 OpenAPI 生成文档FastAPI 自带 Swagger UI不需要额外插件。3.3 项目目录结构建议按下面的结构维护代码和配置open-product-discovery/ ├── server.py # 协议服务端 ├── client.py # 客户端调用示例 ├── schemas.py # 商品和请求响应的数据模型 ├── config.yaml # 服务配置端口、限流、超时 ├── sample_products.json # 测试商品数据 └── tests/ ├── test_search.py ├── test_recommend.py └── test_batch.py这套结构不是强制的但它能让你在后续加功能时不混乱。4. 协议设计与数据格式开放协议的核心是数据格式。如果商品字段不统一AI 就没法稳定理解。下面给出一套示例协议设计命名为 “Open Product Discovery Protocol”简称 OPDP用于演示。4.1 商品对象模型一个最小可用的商品对象应该包含以下字段{ product_id: sku_001, title: 机械键盘 87 键 热插拔, description: 支持热插拔轴体RGB 背光适用于办公和游戏, brand: 示例品牌, categories: [电脑外设, 键盘], attributes: { layout: 87, switch_type: mechanical, connection: usb, color: black }, price: { amount: 299, currency: CNY }, stock_status: in_stock, images: [ { url: https://example.com/img/001.jpg, type: main } ], source_url: https://example.com/products/001, updated_at: 2025-06-01T12:00:00Z }这些字段基本能覆盖搜索、过滤、展示和比较的场景。AI Agent 可以不依赖人类可读页面只凭结构化数据完成判断。4.2 搜索请求格式AI Agent 发起的搜索请求需要包含 query、过滤条件、排序方式、分页信息和返回字段。{ query: 适合程序员用的机械键盘, filters: { brand: [示例品牌], price_min: 100, price_max: 500, currency: CNY, stock_status: in_stock }, sort: { field: relevance, order: desc }, page: 1, page_size: 20, fields: [product_id, title, price, attributes, source_url] }协议里要允许 AI Agent 传自然语言 query也允许传结构化 filters。服务端负责把自然语言 query 转成可执行的检索逻辑。4.3 推荐请求格式推荐和搜索不同它基于用户上下文或种子商品做扩展。{ seed_product_ids: [sku_001], context: { purpose: office, budget: 400 }, strategy: similar, limit: 10 }这里strategy可以是similar、complementary、bestseller等具体由服务端实现。4.4 响应格式统一响应结构方便 AI 解析{ request_id: req_20250701_001, items: [ { product_id: sku_001, score: 0.91, product: { product_id: sku_001, title: 机械键盘 87 键 热插拔, price: { amount: 299, currency: CNY } } } ], pagination: { page: 1, page_size: 20, total: 156 }, trace: { strategy: semantic_search, model: builtin_matcher, latency_ms: 45 } }统一响应结构对 AI 特别重要。Agent 不需要“猜”正确字段路径直接按items[i].product.title读取即可。4.5 批量任务协议批量发现场景下单个请求可能遍历大量商品。建议提供任务式接口而不是同步返回大 JSON。请求{ tasks: [ { task_id: batch_001, queries: [办公键盘, 游戏鼠标], filters: { price_max: 500 } } ] }响应{ batch_id: batch_001, status: accepted, check_url: /v1/batches/batch_001 }客户端通过check_url轮询任务状态拿到结果后再做后续聚合。5. 服务端实现与启动方式下面用一个 FastAPI 示例实现一个简化版协议服务端。这里的目的不是生产代码而是给出启动和测试的最小路径。5.1 数据模型schemas.pyfrom pydantic import BaseModel, Field from typing import Optional from typing import List class Price(BaseModel): amount: float currency: str CNY class Product(BaseModel): product_id: str title: str description: Optional[str] brand: Optional[str] categories: List[str] [] attributes: dict {} price: Price stock_status: str in_stock source_url: Optional[str] class SearchFilter(BaseModel): brand: Optional[List[str]] None price_min: Optional[float] None price_max: Optional[float] None currency: Optional[str] CNY stock_status: Optional[str] None class SearchRequest(BaseModel): query: str Field(..., description自然语言搜索词) filters: Optional[SearchFilter] None page: int 1 page_size: int 20 class SearchResult(BaseModel): product_id: str score: float product: Product class SearchResponse(BaseModel): request_id: str items: List[SearchResult] total: int 05.2 服务端server.py省略真实搜索逻辑用一个简单的关键词匹配 过滤示例。import json import time import uuid from fastapi import FastAPI, HTTPException from schemas import Product, SearchRequest, SearchResponse, SearchResult app FastAPI(titleOpen Product Discovery Protocol, version0.1) # 加载示例商品数据 with open(sample_products.json, r, encodingutf-8) as f: PRODUCTS [Product(**item) for item in json.load(f)] app.get(/health) def health(): return {status: ok} app.post(/v1/search, response_modelSearchResponse) def search(req: SearchRequest): start time.time() # 简易过滤 results [] for p in PRODUCTS: if req.filters: if req.filters.price_min is not None and p.price.amount req.filters.price_min: continue if req.filters.price_max is not None and p.price.amount req.filters.price_max: continue if req.filters.brand and p.brand not in req.filters.brand: continue if req.query and req.query.lower() not in p.title.lower() and req.query.lower() not in p.description.lower(): continue results.append(p) # 分页 start_idx (req.page - 1) * req.page_size page_items results[start_idx:start_idx req.page_size] items [SearchResult(product_idp.product_id, score1.0, productp) for p in page_items] return SearchResponse( request_idfreq_{uuid.uuid4().hex[:12]}, itemsitems, totallen(results) ) if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)5.3 示例商品数据sample_products.json[ { product_id: sku_001, title: 机械键盘 87 键 热插拔, description: 支持热插拔轴体RGB 背光, brand: 示例品牌, categories: [电脑外设, 键盘], attributes: {layout: 87}, price: {amount: 299, currency: CNY}, stock_status: in_stock, source_url: https://example.com/products/001 }, { product_id: sku_002, title: 无线鼠标 静音款, description: 适合办公2.4G 连接, brand: 示例品牌, categories: [电脑外设, 鼠标], attributes: {color: gray}, price: {amount: 99, currency: CNY}, stock_status: in_stock, source_url: https://example.com/products/002 } ]5.4 启动服务python server.py启动后访问健康检查http://127.0.0.1:8000/health接口文档http://127.0.0.1:8000/docs搜索接口http://127.0.0.1:8000/v1/searchFastAPI 自带的 Swagger 文档可以直接做手动测试不需要额外写前端页面。6. 客户端调用与批量任务服务端跑起来后下一步就是让 AI Agent 能调用。下面给出一套通用 Python 客户端示例。6.1 搜索调用import requests url http://127.0.0.1:8000/v1/search payload { query: 键盘, filters: { price_max: 400, stock_status: in_stock }, page: 1, page_size: 10 } resp requests.post(url, jsonpayload, timeout10) data resp.json() print(request_id:, data[request_id]) print(total:, data[total]) for item in data[items]: print(item[product][product_id], item[product][title], item[product][price][amount])预期输出request_id: req_xxxx total: 1 sku_001 机械键盘 87 键 热插拔 299.06.2 批量任务轮询假设服务端实现了/v1/batches接口客户端可以这样设计import time import requests BASE http://127.0.0.1:8000 # 提交批量任务 batch_payload { tasks: [ {task_id: batch_001, queries: [键盘, 鼠标], filters: {price_max: 500}} ] } resp requests.post(f{BASE}/v1/batches, jsonbatch_payload, timeout10) batch resp.json() batch_id batch[batch_id] check_url batch[check_url] # 轮询状态 for _ in range(30): status_resp requests.get(f{BASE}{check_url}, timeout10).json() if status_resp[status] completed: for task in status_resp[results]: print(task[task_id], len(task[items]), items) break time.sleep(2) else: print(timeout: 任务未完成)批量任务的核心是“提交 - 轮询 - 拿到结果”。不要让 Agent 长时间挂在一个 HTTP 请求上等结果。6.3 批量任务设计建议每个任务使用独立 task_id方便日志追踪。任务结果落盘到对象存储或数据库而不是保存在内存。设置最大重试次数避免失败任务无限重试。批量任务要有去重逻辑防止重复推荐同一商品。如果处理超时服务端应该返回202 Accepted而不是阻塞在200。7. 功能测试与效果验证拿到一个协议服务验证时不要只看“能返回数据”要按下面的测试维度逐项检查。7.1 健康检查与文档测试目的确认服务正常运行。操作curl http://127.0.0.1:8000/health预期结果{status:ok}7.2 基础搜索测试测试目的验证自然语言搜索和过滤条件是否生效。请求curl -X POST http://127.0.0.1:8000/v1/search \ -H Content-Type: application/json \ -d {query:键盘,filters:{price_max:400}}判断标准返回total大于 0。每个items中的商品标题包含“键盘”或描述相关。商品价格不超过 400。7.3 空结果测试测试目的检查空结果时响应是否正常。请求curl -X POST http://127.0.0.1:8000/v1/search \ -H Content-Type: application/json \ -d {query:不存在的商品名称}预期结果返回total0items[]HTTP 状态码仍然是 200。不能返回 500。7.4 分页测试测试目的验证分页稳定。第 1 页page_size1和第 2 页page_size1不能有重复商品。页码超出范围时返回空列表而不是崩溃。7.5 批量任务测试测试目的验证异步任务流程。提交包含多个 query 的批量任务。轮询check_url确认任务状态最终为completed。检查每个 task 的结果数量和内容是否符合预期。7.6 失败场景注入测试目的确认服务在异常数据下表现稳定。发送price_max为字符串而不是数字。发送缺失query字段的请求。发送超大page_size例如 99999。良好设计下服务端应该返回 422 参数校验错误而不是堆栈异常。7.7 常见失败原因测试现象可能原因排查方式返回 404路由路径错误检查服务端路由注册返回 422请求体字段类型或缺失查看/docs里的 schema返回 500服务端未捕获异常查看 uvicorn 控制台日志返回空结果数据未加载或过滤条件过严检查sample_products.json和搜索逻辑批量任务一直pending异步执行器没跑检查是否启动了后台 worker8. 资源占用与性能观察这个协议服务不涉及 GPU 推理资源占用集中在 CPU 和内存。但如果是真实生产环境仍需要关注性能。8.1 观察指标请求平均延迟建议看 p95而不是平均值。内存占用加载的商品对象数量决定内存量。CPU 使用率搜索逻辑越复杂CPU 占用越高。端口状态确认 8000 端口没有被其他进程占用。8.2 如何测试用简单的并发请求脚本观察import time import requests from concurrent.futures import ThreadPoolExecutor url http://127.0.0.1:8000/v1/search payload {query: 键盘, page_size: 5} def call(_): start time.time() r requests.post(url, jsonpayload, timeout10) return time.time() - start with ThreadPoolExecutor(max_workers20) as ex: latencies list(ex.map(call, range(100))) print(avg latency:, sum(latencies) / len(latencies)) print(max latency:, max(latencies))注意不要用高并发去压本地小服务可能会触发端口耗尽或进程崩溃。8.3 性能优化方向商品数据量大时使用 Redis 或内存索引缓存。搜索接口加参数校验和限流防止被刷。将自然语言 query 转语义向量的部分独立成模型服务不要阻塞主接口。批量任务用消息队列消化而不是在 HTTP worker 里同步执行。8.4 显存思考这个主题不涉及显存。如果你的协议服务内部集成了向量模型做语义搜索那么模型推理部分会有显存需求。但协议层本身没有任何显存占用。建议在架构上把“语义模型推理”和“协议服务”拆开部署协议层保持轻量。9. 常见问题与排查方法下面汇总基于这套通用协议实现中最容易遇到的问题以及处理思路。问题现象可能原因排查方式解决方案依赖安装失败Python 版本不匹配或缺少构建工具检查pip install日志升级 Python 或使用虚拟环境启动后接口文档打不开端口被占用lsof -i :8000或netstat -ano更换--port 8001搜索返回乱序没有定义相关性排序检查返回score值实现排序逻辑或默认按相关度商品数据重复商品 ID 不唯一检查 JSON 数据的product_id导入前做去重批量任务卡在 pending后台 worker 未启动查看任务队列消费者日志启动 worker 增加日志API 调用超时搜索逻辑太慢打印 SQL 或搜索耗时加索引或缓存接口返回 500数据中包含残缺字段检查 pydantic 校验报错修正数据或增加默认值AI Agent 解析失败响应格式不稳定检查request_id和items是否恒定固定 JSON 结构避免字段值类型变化10. 最佳实践与使用建议10.1 先小批量验证协议语义第一次接入时不要直接对接全量商品库。准备 100 个左右覆盖不同类的测试商品先把协议跑通再逐步接入真实数据。这样能快速发现字段映射和查询逻辑的问题。10.2 商品数据要有明确来源与版本协议里的每个商品对象最好带上source_url和updated_at否则 AI Agent 拿到结果后无法溯源。对于比价、推荐场景数据来源和更新时间直接影响结果可信度。10.3 为不同调用方设置不同权限开放协议不等于完全公开。建议至少设置三种角色只读搜索可以查商品但不能看成本价格。批量任务可以发起批量查询但受限频限制。管理端可以维护商品数据。权限通过 ApiKey 和 scope 字段控制代码里校验API_KEYS { test_key: {scopes: [search, batch]} } def check_scope(key, required_scope): if key not in API_KEYS: return False return required_scope in API_KEYS[key][scopes]10.4 日志里不要记录用户隐私协议接口本身不要求传用户身份。如果上游把用户 ID 放在 context 参数里日志输出时要脱敏。不要打印完整用户 id、手机号、地址等字段。10.5 为 AI Agent 设计容错策略AI Agent 调用接口时可能遇到 429 限流、500 错误、网络超时。客户端要做好指数退避重试并缓存上一次成功结果。避免请求失败时直接返回空推荐给用户。def call_with_retry(url, payload, retries3): for i in range(retries): try: resp requests.post(url, jsonpayload, timeout5) resp.raise_for_status() return resp.json() except Exception: time.sleep(2 ** i) return None10.6 版权与合规必须在协议层声明如果你是商品数据提供方建议在协议响应中加入license字段标明数据使用范围。如果协议用于跨境或跨平台商品比较需要先确认当地法律和各平台条款。涉及品牌商标、博主推荐、用户生成内容时必须获得授权。10.7 发布前做一轮 AI 结果抽查AI 产品发现经常会放大错误一个商品标题写错就可能被模型推荐给大量用户。上线前用固定 query 列表跑一遍回归测试记录每次推荐的商品 ID 和分值。发现异常数据及时下架不要指望模型自动纠错。11. 总结与下一步开放协议的核心价值是“让 AI 不再为每个平台手写适配器”。通过统一商品模型、搜索接口、推荐响应和批量任务结构AI Agent 可以把精力放在理解用户意图和优化推荐策略上而不是纠结字段名和接口差异。如果你准备在自己的项目里落地这套思路建议先做三件事定义一份最小商品 JSON Schema只包含你业务里真正用到的字段。写一个只有搜索和详情两个端点的 MVP 服务用 FastAPI 跑通。用一个模拟 Agent 客户端连续调用 50 次记录失败率和返回字段稳定性。最容易踩的坑也提前说清楚一是商品数据字段随意变化导致 Agent 解析报错二是批量任务同步化导致请求超时三是不设权限把商品库直接暴露给公网。后续如果要扩展可以往这几个方向走引入向量检索做语义搜索为每个商品生成 embedding增加订阅推送让 Agent 感知库存和价格变化支持多语言商品描述把协议层做成插件接入现有的 MCPModel Context Protocol或第三方 Agent 框架。协议本身不是银弹但它能把 AI 产品发现的协作边界定清楚。先从一个搜索接口开始把一个字段一个请求打磨稳定比一开始就铺大而全的开放平台更靠谱。建议收藏这篇文章部署时对照检查。