尧图网站建设 
“你们的产品目录能不能直接对接”——这是过去半年里我被问到最多的一句话。当一个 AI Agent 要帮用户完成“从模糊想法到具体产品”的发现过程时最先卡住的往往不是模型能力而是各家平台 API 的差异。搜索接口、筛选参数、结果结构、价格字段、口径含义都各不相同AI Agent 每接一个数据源就要写一遍新逻辑既难维护也很难快速扩展。更好的做法不是继续给 AI Agent 增加“适配器”而是先定义一套开放、统一、可扩展的协议让所有 Provider 按同一套规则提供服务。这篇文章会从协议设计、消息模型、最小服务端实现、客户端接入、安全边界和常见问题六个方面完整拆解“AI 中介产品发现”这一场景怎么落地。无论你是后端架构师、AI 应用开发者还是正在做 agent/hub 类产品的技术负责人都适合沿着这篇文章的思路从零搭一套自己的发现协议。1. 为什么需要“AI 中介产品发现”开放协议1.1 什么是 AI 中介的产品发现产品发现Product Discovery这个词听起来很大但在实际业务里非常具体。传统电商搜索是用户在搜索框里输入“降噪耳机”然后浏览器根据关键词返回商品列表。这个过程中用户是主动发起人平台负责把搜索结果按某个规则排序。而 AI 中介的产品发现则不同用户可能只对 AI Agent 说“我每天通勤两小时偶尔还要出差想要一款续航不错、不要超过一千块的耳机”。AI Agent 需要先理解这句话里的预算、场景、品类、约束条件再把这个需求转成结构化的搜索条件最终从多个商品目录中找到合适产品。整个过程里AI Agent 是“用户和商品目录之间”的中介。这种模式对协议有天然要求。用户的需求不会是一句“耳机”可能是多轮对话之后才逐渐清晰。因此 AI Agent 需要把“上下文”传递给商品目录服务而不是只传一个关键词。同时AI Agent 同时对接的目录可能有很多个有的提供硬件产品有的提供 SaaS 服务有的提供内容课程。如果每个目录都用自己的字段结构、筛选语法、分页方式和错误码AI Agent 的对接成本会以指数级上升。所以我们需要一个开放协议把这套“发现能力”通用化。1.2 当前各平台 API 对接的痛点现在大部分平台都开放了 HTTP API但真正去对接时问题非常多。首先是搜索参数不统一有的平台用q表示关键词有的用keyword还有的用searchText价格筛选有的用price_min/price_max有的用range100-1000。其次是结果结构差异大有的返回data.goodsList有的返回products.items价格字段有的是字符串有的是浮点有的把金额和币种拆成两个字段。AI Agent 每次接入一个新数据源都需要写一层“字段映射”然后不断处理边界情况。更麻烦的是上下文无法传递。AI Agent 多轮对话中已经知道用户预算 1000 元以内、只需要通勤使用但平台的“搜索”接口根本不接受这类上下文。要么我们只能把用户意图压缩成关键词丢掉大量可用信息要么我们自己在 Agent 侧做过滤把平台结果拿回来再筛一遍。这个过程中还会出现重复排序、分页错乱、召回不完整等问题。协议缺失导致 AI 产品发现的质量始终不稳定。1.3 开放协议到底要解决什么开放协议要解决的核心问题是让“AI Agent”和“产品发现服务”之间拥有稳定的通信语言。协议要定义清楚三件事第一消息格式也就是请求和响应用什么结构表达第二行为语义比如搜索、比较、排序、上下文续传、事件上报各自应该怎样工作第三非功能性要求比如安全认证、错误处理、限流、审计日志。只要 Provider 遵循这套协议AI Agent 就不需要关心底层数据存在哪、用什么语言开发、排序算法是什么。协议还有一个容易被忽略的价值把“产品发现”从单一平台 API 的绑定中解放出来。企业可以把多个自营平台、第三方供应商的产品目录通过同一个网关做接入让 AI Agent 统一调度。对用户而言获得的是更完整的“发现体验”对企业而言增加一个可被 AI Agent 发现的渠道会比单独开发一个 App 或小程序更容易触达用户。正因如此“开放”这两个字很重要协议不应该是某家公司私有标准而应该公开字段定义、允许任何人实现 Provider 或客户端。1.4 协议适用场景我建议不要一上来就追求大而全的协议而是先覆盖高频场景。最核心的接口有三个搜索search、比较compare、事件上报event。搜索解决“从模糊需求到候选集合”比较解决“在候选里做多商品属性对比”事件上报解决“用户最终的选择和反馈回传”。先把这个闭环跑通再逐步扩展推荐、收藏、订阅等能力。本文的完整示例就以这三个场景为核心展开但代码部分会重点实现最常用到的搜索接口另外两个会给出清晰的扩展建议。2. 协议设计参与者、消息模型与核心流程2.1 参与者划分协议里至少包含三类参与者。第一类是 AI Agent也就是调用方它负责理解用户需求、维护对话状态、整合多个 Provider 的返回结果。第二类是 Discovery Provider也就是产品发现能力提供方它可以是电商平台、SaaS 目录、企业内部的商品中心、甚至是第三方数据服务商。第三类是 Identity Provider负责认证和权限管理通常与 Auth Provider 一起工作。在最小实现里Identity 可以通过 API Key 或 OAuth2 Bearer Token 完成不必单独部署服务。AI Agent 和 Discovery Provider 之间是“调用与被调用”的关系但这个关系不能是不受约束的。协议要求每个请求必须有唯一 request_id方便链路追踪每个 Provider 的返回必须携带 search_id便于后续事件上报时关联“第几次搜索”。这两个标识是很多接入方案容易忽略的细节但对线上问题排查至关重要。2.2 消息模型定义为了让不同系统之间通信不产生歧义协议需要用 JSON Schema 或类型定义把消息结构固下来。下面是一份简化但可落地的消息模型。SearchRequest是 AI Agent 发给 Provider 的搜索请求字段包括 request_id、query、context、limit、offset 和一些自定义扩展字段。ProductCard是产品信息的最小载体字段包括 product_id、title、description、price、attributes、source 和 score。attrubutes 里可以放租户自定义属性比如“降噪等级”“续航时长”“发货地区”。{ request_id: req-20250101-001, query: 降噪耳机, context: { user_scene: commute, budget: { currency: CNY, max: 1000 }, preferences: { brands: [demo-a, demo-b] } }, limit: 10, offset: 0, extensions: { sort: score } }响应里除了商品列表还需要返回 total 总数方便 Agent 判断是否要继续翻页同时返回 search_id由 Provider 生成用于关联后续上报事件。这里的 context 可以是半结构化数据但协议会建议标准化几个常见字段user_scene、budget、preferences其他业务字段放入extensions这样既保证通用性又保留扩展空间。{ request_id: req-20250101-001, code: 0, message: ok, search_id: search-xxxx, total: 23, items: [ { product_id: p-001, title: 无线蓝牙降噪耳机 Pro, description: 适合通勤与办公支持主动降噪续航 20 小时。, price: { currency: CNY, amount: 399.00 }, attributes: { category: audio, brand: demo-a, noise_canceling: true }, source: demo-a, score: 0.96 } ] }2.3 核心请求链路一次完整的 AI 中介产品发现通常有五个步骤。第一步用户用自然语言描述需求AI Agent 对需求做意图识别和结构化抽取。第二步AI Agent 根据对话历史构造一个或多个SearchRequest调用一个或多个 Provider。第三步Provider 根据 query 和 context 返回召回结果Agent 对结果做去重、合并、排序。第四步AI Agent 把结果生成推荐文案或商品列表展示给用户。第五步用户做出选择后Agent 调用event接口回传偏好和最终选择帮助 Provider 做后续优化。从协议角度看第二步到第五步都是标准的 HTTP 调用区别只在消息结构。整个链路里最容易出问题的点是第二步到第三步AI Agent 传入的 context 是否真的被 Provider 用来参与召回而不是被忽略。因此协议不只是定义字段还要约定 Provider 应该在 sorting、filtering、ranking 中如何使用这些字段。如果 Provider 不理解某个 context 字段应该明确忽略并在 response 里给出warnings提示而不是让 Agent 误以为自己的约束已经生效。这一点会在后面实战代码里体现。3. 环境准备与最小服务端实现3.1 环境说明为了不把篇幅浪费在某个具体框架的版本上这里用一个非常轻量级的 Python 技术栈做演示。操作系统不限Windows、macOS、Linux 都可以。本地需要安装 Python 3.10 或更高版本然后安装 FastAPI 和 Uvicorn。FastAPI 负责 HTTP API 的定义与参数校验Uvicorn 负责启动本地开发服务器。如果你更习惯 Node.js 或 Go协议本身并不限制语言核心是消息结构和行为语义保持一致代码实现思路可以直接迁移。实战项目的目录结构如下建议直接在本地新建一个aipd-demo文件夹aipd-demo/ ├── app/ │ ├── __init__.py │ ├── schema.py # 协议消息模型 │ ├── store.py # 内存商品数据源 │ └── main.py # FastAPI 服务入口 ├── requirements.txt # 依赖清单 └── client_demo.py # AI Agent 客户端调用示例3.2 安装依赖在aipd-demo目录下创建requirements.txt写入下面几行依赖。版本号建议以你本地实际情况为准如果只需要跑通示例直接安装最新版本通常没有问题。fastapi uvicorn pydantic requests然后执行安装命令pip install -r requirements.txt如果你使用虚拟环境建议先创建并激活虚拟环境再安装依赖。这里不做强制要求但线上开发时强烈建议使用虚拟环境或容器避免依赖污染。3.3 定义协议消息模型在app/schema.py中我们使用 Pydantic 把上一节的 JSON 消息模型变成 Python 类。这样 FastAPI 会自动帮我们完成请求体校验和响应序列化也能在开发阶段获得类型提示。# app/schema.py from typing import List, Optional, Dict, Any from pydantic import BaseModel, Field class Context(BaseModel): AI Agent 传过来的半结构化上下文 user_scene: Optional[str] Field(defaultNone, description用户场景例如 commute、gift) budget: Optional[Dict[str, Any]] Field(defaultNone, description预算例如 {currency: CNY, max: 1000}) preferences: Optional[Dict[str, Any]] Field(defaultNone, description偏好例如品牌、标签) extensions: Optional[Dict[str, Any]] Field(defaultNone, description自定义扩展字段) class SearchRequest(BaseModel): request_id: str Field(..., description调用方生成唯一请求ID) query: str Field(..., description用户查询关键词) context: Optional[Context] Field(defaultNone, description多轮对话上下文) limit: int Field(default10, ge1, le50, description返回条数) offset: int Field(default0, ge0, description分页偏移) extensions: Optional[Dict[str, Any]] Field(defaultNone, description扩展参数) class ProductCard(BaseModel): product_id: str title: str description: str price: Dict[str, Any] Field(..., description价格必须包含 currency 和 amount) attributes: Dict[str, Any] Field(default_factorydict) source: str Field(..., description数据来源标识) score: Optional[float] Field(defaultNone, description相关性分数) class SearchResponse(BaseModel): request_id: str code: int Field(default0, description0 表示成功) message: str Field(defaultok) search_id: str Field(default, descriptionProvider 生成的搜索ID) total: int Field(default0, description满足条件的总数) items: List[ProductCard] Field(default_factorylist, description商品列表) warnings: List[str] Field(default_factorylist, description协议层告警信息)这里有一个细节值得说明price被定义成Dict[str, Any]而不是两个独立字段。原因是在 AI 发现场景里不同供应商的价格结构可能不同有的会包含currency、amount有的还会带original_amount或tax。协议要求至少包含currency和amount额外字段则保留在字典里这样兼容性会更好。3.4 编写内存商品数据源接着在app/store.py中模拟一小部分商品数据。真实项目中这里可以替换成数据库查询、搜索服务调用甚至其他 Provider 的 HTTP 调用。为了让示例专注协议我们使用列表存储。# app/store.py from typing import List, Tuple PRODUCTS [ { id: p-001, title: 无线蓝牙降噪耳机 Pro, description: 适合通勤和办公支持主动降噪续航 20 小时。, price: {currency: CNY, amount: 399.00}, attrs: {category: audio, brand: demo-a, noise_canceling: True}, source: demo-a, }, { id: p-002, title: 头戴式降噪耳机 Max, description: 长续航头戴式耳机适合出差和安静办公。, price: {currency: CNY, amount: 899.00}, attrs: {category: audio, brand: demo-b, noise_canceling: True}, source: demo-b, }, { id: p-003, title: 便携蓝牙音箱 Mini, description: 小体积、防水设计适合户外使用。, price: {currency: CNY, amount: 199.00}, attrs: {category: audio, brand: demo-a, noise_canceling: False}, source: demo-a, }, ] def local_search(query: str, limit: int 10, offset: int 0) - Tuple[List[dict], int]: 基于关键词和描述做一次非常简单的本地搜索 q query.strip().lower() if not q: return [], 0 matched [ product for product in PRODUCTS if q in product[title].lower() or q in product[description].lower() ] return matched[offset:offset limit], len(matched)注意local_search返回两个值当前页数据和匹配总数。分页逻辑很简单但如果接入真实搜索服务排序和过滤应该下沉到存储层而不是在 Python 里硬过滤。3.5 实现 FastAPI 服务在app/main.py中实现协议的服务端。第一步是创建 FastAPI 实例第二步是定义/v1/search接口接收SearchRequest第三步是调用local_search并组装响应。# app/main.py import uuid from fastapi import FastAPI, HTTPException from app.schema import SearchRequest, SearchResponse, ProductCard from app.store import local_search app FastAPI( titleAIPD Provider Demo, version0.1.0, descriptionAIPD (AI-mediated Product Discovery) 协议最小服务端示例, ) app.post(/v1/search, response_modelSearchResponse) async def search(req: SearchRequest): try: matched, total local_search( queryreq.query, limitreq.limit, offsetreq.offset, ) items [ ProductCard( product_idproduct[id], titleproduct[title], descriptionproduct[description], priceproduct[price], attributesproduct[attrs], sourceproduct[source], ) for product in matched ] warnings [] if req.context and req.context.budget: # 示例中不真正实现预算过滤但通过 warning 提示调用方 warnings.append(budget filter is not enforced in demo provider) return SearchResponse( request_idreq.request_id, search_idstr(uuid.uuid4()), totaltotal, itemsitems, warningswarnings, ) except Exception as exc: raise HTTPException(status_code500, detailstr(exc))这段代码里有一个非常实用的协议设计当 Provider 不能处理 context 中的预算条件时不要静默忽略而是通过warnings字段告诉调用方。AI Agent 看到 warning 后可以在文案里主动提示用户“当前搜索没有严格按预算过滤”避免用户产生误解。这个细节会让 AI 产品发现的可信度大幅提升。4. 客户端接入与实战验证4.1 启动服务端进入aipd-demo目录后执行下面的命令启动本地服务uvicorn app.main:app --reload --port 8000启动成功后终端会显示 FastAPI 的访问地址。打开http://localhost:8000/docs你会看到 Swagger 文档页面可以直接在页面上测试/v1/search接口。这个内置文档在开发阶段非常方便也方便前端同学快速理解协议。4.2 使用 curl 快速验证在不写代码的情况下我们可以先用curl调用服务。下面的命令发送一个最简单的搜索请求查询“降噪耳机”。curl -X POST http://localhost:8000/v1/search \ -H Content-Type: application/json \ -d { request_id: req-curl-001, query: 降噪耳机, limit: 5 }预期返回中会看到两条商品记录p-001 和 p-002。因为它们都包含“降噪”和“耳机”关键词。同时返回的search_id是一个随机生成的 UUIDtotal是 2warnings为空。如果你把 query 改成“蓝牙”返回结果会同时包含 p-001 和 p-003因为 p-003 的描述里包含“蓝牙”和“音箱”。4.3 编写 AI Agent 客户端调用在实际的 AI Agent 工程中客户端通常不会直接暴露给用户而是在 Agent 的服务端完成调用。这里编写client_demo.py模拟 AI Agent 内部的联网函数。它负责构造SearchRequest然后解析SearchResponse最后把结果整理成用户可读的文本。# client_demo.py import requests def search_products(base_url: str, query: str, request_id: str): payload { request_id: request_id, query: query, limit: 5, context: { user_scene: commute, budget: {currency: CNY, max: 1000}, }, } resp requests.post(f{base_url}/v1/search, jsonpayload, timeout10) resp.raise_for_status() return resp.json() def format_result(data: dict) - str: lines [] if data.get(warnings): lines.append(注意) for warning in data[warnings]: lines.append(f- {warning}) lines.append(推荐商品) for item in data.get(items, []): price item.get(price, {}) amount price.get(amount, 0) currency price.get(currency, CNY) lines.append(f- {item[title]} ({currency} {amount:.2f})) lines.append(f共找到 {data.get(total, 0)} 件商品) return \n.join(lines) if __name__ __main__: base_url http://localhost:8000 result search_products(base_url, 降噪耳机, req-client-001) print(format_result(result))运行客户端python client_demo.py预期输出仅供参考推荐商品 - 无线蓝牙降噪耳机 Pro (CNY 399.00) - 头戴式降噪耳机 Max (CNY 899.00) 共找到 2 件商品由于我们的 demo Provider 没有真正实现预算过滤所以这里没有输出 warning。如果后面改造成真实数据源并且不支持 context 里的某个字段客户端就会看到 warning。AI Agent 可以根据 warning 决定如何向用户表达这是协议带来的弹性空间。4.4 结果说明与协议行为验证从上面的完整调用可以看出协议的价值在于“两端只依赖同一个消息契约”。AI Agent 端不需要知道 Provider 的数据库结构只需要解析SearchResponseProvider 端不需要关心 Agent 是用了 GPT、Claude 还是本地模型只需要接收SearchRequest。如果要新增一个 Provider只需要按协议实现/v1/search再在 Agent 端的 Provider 列表里登记即可不需要修改 Agent 的消息处理逻辑。反过来如果 Agent 要从“只能搜商品”扩展到“能比较商品”只需要在协议里增加一个/v1/compare端点所有遵守协议的 Provider 都能在同一时间拥有这个能力。5. 安全边界、协议扩展与生产落地5.1 鉴权与最小权限在本地 demo 里接口完全公开任何人都可以调用。生产环境则必须加上身份认证和权限控制。建议使用 API Key 或 OAuth2 Bearer Token 作为调用凭证。每个 AI Agent 只能访问它有权限看到的商品目录并且要对每一次调用做审计。这里需要特别强调最小权限原则AI Agent 的接口令牌不应该拥有修改商品、删除商品、管理用户的权限它只需要aipd:search和aipd:event:write这两类核心权限。在 FastAPI 中可以通过依赖注入轻松实现 API Key 校验。这里给出一个最小示例片段。注意真正生产环境应该使用更安全的密钥存储和轮换机制。# app/auth.py最小鉴权示例 from fastapi import Header, HTTPException API_KEYS { demo-agent: sk-demo-123456, } def verify_api_key(x_api_key: str Header(...)): if x_api_key not in API_KEYS.values(): raise HTTPException(status_code401, detailinvalid api key)然后修改接口定义加入dependencies[Depends(verify_api_key)]。这样客户端调用时就必须携带X-API-Key: sk-demo-123456。关于密钥管理、轮换和最小权限需要结合你实际的 IAM 体系来做协议本身不限定具体实现。5.2 分页、排序与结果稳定性AI 产品发现场景里分页设计比普通 API 要更谨慎。用户在多轮对话中可能多次翻页如果每次翻页时 Provider 内部排序不稳定就会出现重复商品或漏掉商品。因此协议要求 Provider 对同一 query 的排序结果是确定性的。如果排序依赖模型分数建议在响应里返回score并尽量使用tiebreaker字段保证稳定。offset/limit是兼容 REST 习惯的简单方案但数据量很大时性能会变差更推荐游标分页。可以在协议的extensions字段里加入cursor参数用于替代 offset这个扩展不影响核心字段。同时AI Agent 侧也要做去重。多个 Provider 可能返回同一件商品比如同一家店的商品既在自营目录里也在聚合平台里此时source字段和product_id组合是最好的去重键。Agent 在合并结果时可以根据score重新排序也可以保留 Provider 自身的顺序关键是在协议层把这些规则说清楚。5.3 上下文传递的实用策略协议里的context是 AI Agent 与 Provider 之间传递“用户软约束”的主要手段。但注意context 不是无限大的对话历史更不是 Prompt 原文。推荐的做法是AI Agent 在调用 Provider 前先把用户需求做一次结构化抽取只把与商品筛选相关的字段放进 context。例如user_scene、budget、brands、delivery_area这些字段可以标准化至于用户说过的“心情不好”“喜欢简约风”之类的信息如果 Provider 支持语义理解可以放进extensions否则不要强行传递。在 demo 服务里我们特意返回了 budget warning就是为了向 Agent 说明“预算约束没有被执行”。在真实协议里一个更负责的 Provider 应该在服务端实现 budget 过滤而不是把过滤责任推给 Agent。如果暂时做不到就要通过 warning 或显式字段告诉 Agent 哪些 context 字段被忽略了这是协议比普通 REST API 更高级的地方。5.4 版本管理与兼容演进开放协议发布后一定会遇到版本演进。最简单的方式是在 URL 路径中加入版本号例如/v1/search、/v2/search。版本的语义要提前约定好新增字段可以向后兼容修改字段含义或删除字段属于不兼容变更必须升大版本。建议每个字段都使用 optional 方式设计这样新版本可以在旧版本基础上增加信息而不会破坏已有调用方。除了 URL 版本号还建议在响应头中返回协议版本号例如X-AIPD-Version: 1.0。这样 AI Agent 可以快速判断是否调用了旧版本 Provider并做相应适配。如果你把协议发布到公司内部网关还可以结合多环境部署和灰度发布来控制不同版本的上线范围。5.5 限流、超时与降级AI Agent 调用 Provider 的链路比普通前端调用更长一个 Agent 往往要同时调用多个 Provider。为了避免某个慢 Provider 拖垮整体体验客户端必须设置合理的超时时间建议默认 2-3 秒。如果某个 Provider 超时AI Agent 应该自动降级跳过该 Provider不要把错误直接暴露给用户。Provider 端则要提供容量保护和限流避免单个 Agent 的异常重试影响其他调用方。协议里可以增加一个统一的错误响应结构例如error字段包含code、message、details。客户端遇到统一错误结构时可以快速归类错误遇到非标准错误时只能做通用兜底。为了让错误更容易排查每个日志都必须带上request_id和search_id形成完整追踪链。6. 常见问题与排查思路下面整理 AI 中介产品发现协议在落地过程中最常见的几类问题。问题现象常见原因解决思路调用/v1/search返回 401API Key 缺失或错误检查请求头X-API-Key确认令牌有效期和权限范围接口返回 404路径版本号不匹配确认客户端与 Provider 版本一致例如/v1/search响应解析失败Provider 返回结构与协议不一致先用docs接口或 Schema 校验抓包比对字段名和类型商品结果重复多个 Provider 返回同一商品使用source product_id去重再按 score 重新排序返回结果为空但 total 较大分页 offset 计算错误或游标过期检查分页参数推荐使用 cursor 分页context 里的预算没生效Provider 未实现该过滤逻辑检查响应warnings字段联系 Provider 服务方实现请求超时Provider 响应慢或网络链路长客户端缩短超时时间开启降级重试策略多轮对话后结果漂移Provider 排序不稳定要求 Provider 提供稳定排序并返回score日志里查不到问题request_id 没有贯穿全链路通信和日志都统一透传 request_id这些问题的根源大多不是某一个代码 bug而是“没有协议意识”。例如返回结构不一致、版本混乱、上下文语义模糊都是可以先通过协议定义规避的。排错时第一步永远先确认消息结构是否符合协议第二步看有没有warnings或error信息第三步再检查具体业务逻辑。不要一上来就去查数据库或改代码那样效率很低。7. 最佳实践与工程建议7.1 先定契约再写代码在正式开发前先确定一份“最小可用协议”。可以把SearchRequest和SearchResponse的多数字段先定义成 optional然后让两端以这份契约为准进行开发。不要一边写代码一边改字段否则很容易出现 Provider 和 Agent 各维护一套“隐式约定”的情况。建议把协议 Schema 单独做成一个共享包服务端和客户端都从同一个包引用从源头避免字段不一致。7.2 协议的每一层都要可观测AI 中介产品发现链路里用户最终看到的只是一个推荐列表但中间经过了“意图理解 - 调用多Provider - 结果融合 - 文案生成”等多个环节。任何一个环节出问题都会让结果质量下降。因此要在协议层增加可观测性。建议每个 Provider 在响应头里返回X-Request-Id、X-Search-Id、X-Duration-MsAgent 端统一打印到日志。配合链路追踪系统可以让“哪个 Provider 慢”“哪个 Provider 结果差”一目了然。7.3 关注用户授权与隐私边界协议传递 context 时要特别小心隐私信息。用户位置、年龄、收入等敏感信息不能随意传给第三方 Provider必须脱敏或只传化名 ID。每一类 Provider 能拿到哪些字段需要在授权协议里写清楚。AI Agent 在调用 Provider 之前要主动检查当前用户是否已经授权该数据源未授权时跳过调用。最小权限原则同样适用于数据字段不要把所有上下文一股脑传给所有 Provider。7.4 结果要做解释而不仅是给出列表AI 产品发现与普通搜索不同的地方在于用户需要“为什么推荐这个”。协议里的score字段只能表达相关性不足以说明理由。建议在extensions中补充reasons数组例如“符合预算上限”“匹配品牌偏好”“用户评分高”。Provider 可以提供这些理由AI Agent 将它们组合成推荐文案。这个设计会在用户体验上有明显提升也是 AI 产品发现相对传统搜索的增量价值。7.5 让协议的进化沿着真实场景走不要一开始就设计出功能繁多的协议。正确的做法是先跑通“搜索到展示到反馈”的最小闭环然后在真实业务中观察哪些字段缺失、哪些流程无法覆盖。比如对比功能一开始可能不需要但用户经常在几件商品之间犹豫那就应该优先补/v1/compare接口。协议应该以迭代方式演进每次升级都要有日志记录和兼容性说明。8. 总结与下一步这篇文章从实际业务痛点出发完整讨论了“AI 中介产品发现”为什么需要开放协议并围绕SearchRequest、SearchResponse、ProductCard等核心消息模型设计了一套最小可落地的协议。我们用 FastAPI 实现了一个简单的 Provider 服务端并编写了 AI Agent 客户端调用示例同时补充了鉴权、分页、上下文传递、版本演进和可观测性等生产级细节。虽然示例的服务端非常简单但它已经构成了一个完整的协议闭环只要你愿意完全可以照着这套思路把内部商品中心、第三方供应商、甚至其他 Agent 服务都接入进来。下一步建议你先在自己的项目里定义一套最小协议不一定要做得很完整先把“搜索-比较-反馈”这条主线跑通。然后从真实用户反馈里观察协议缺了什么、哪里语义不清楚再按版本机制逐步演进。开放协议不是一次性的架构设计而是持续演进的一套协同规则。如果你正在做 AI Agent 或产品目录相关的项目可以从今天的 /v1/search 开始迈出第一步。 AI中介产品发现开放协议：从API适配到统一搜索实战
“你们的产品目录能不能直接对接”——这是过去半年里我被问到最多的一句话。当一个 AI Agent 要帮用户完成“从模糊想法到具体产品”的发现过程时最先卡住的往往不是模型能力而是各家平台 API 的差异。搜索接口、筛选参数、结果结构、价格字段、口径含义都各不相同AI Agent 每接一个数据源就要写一遍新逻辑既难维护也很难快速扩展。更好的做法不是继续给 AI Agent 增加“适配器”而是先定义一套开放、统一、可扩展的协议让所有 Provider 按同一套规则提供服务。这篇文章会从协议设计、消息模型、最小服务端实现、客户端接入、安全边界和常见问题六个方面完整拆解“AI 中介产品发现”这一场景怎么落地。无论你是后端架构师、AI 应用开发者还是正在做 agent/hub 类产品的技术负责人都适合沿着这篇文章的思路从零搭一套自己的发现协议。1. 为什么需要“AI 中介产品发现”开放协议1.1 什么是 AI 中介的产品发现产品发现Product Discovery这个词听起来很大但在实际业务里非常具体。传统电商搜索是用户在搜索框里输入“降噪耳机”然后浏览器根据关键词返回商品列表。这个过程中用户是主动发起人平台负责把搜索结果按某个规则排序。而 AI 中介的产品发现则不同用户可能只对 AI Agent 说“我每天通勤两小时偶尔还要出差想要一款续航不错、不要超过一千块的耳机”。AI Agent 需要先理解这句话里的预算、场景、品类、约束条件再把这个需求转成结构化的搜索条件最终从多个商品目录中找到合适产品。整个过程里AI Agent 是“用户和商品目录之间”的中介。这种模式对协议有天然要求。用户的需求不会是一句“耳机”可能是多轮对话之后才逐渐清晰。因此 AI Agent 需要把“上下文”传递给商品目录服务而不是只传一个关键词。同时AI Agent 同时对接的目录可能有很多个有的提供硬件产品有的提供 SaaS 服务有的提供内容课程。如果每个目录都用自己的字段结构、筛选语法、分页方式和错误码AI Agent 的对接成本会以指数级上升。所以我们需要一个开放协议把这套“发现能力”通用化。1.2 当前各平台 API 对接的痛点现在大部分平台都开放了 HTTP API但真正去对接时问题非常多。首先是搜索参数不统一有的平台用q表示关键词有的用keyword还有的用searchText价格筛选有的用price_min/price_max有的用range100-1000。其次是结果结构差异大有的返回data.goodsList有的返回products.items价格字段有的是字符串有的是浮点有的把金额和币种拆成两个字段。AI Agent 每次接入一个新数据源都需要写一层“字段映射”然后不断处理边界情况。更麻烦的是上下文无法传递。AI Agent 多轮对话中已经知道用户预算 1000 元以内、只需要通勤使用但平台的“搜索”接口根本不接受这类上下文。要么我们只能把用户意图压缩成关键词丢掉大量可用信息要么我们自己在 Agent 侧做过滤把平台结果拿回来再筛一遍。这个过程中还会出现重复排序、分页错乱、召回不完整等问题。协议缺失导致 AI 产品发现的质量始终不稳定。1.3 开放协议到底要解决什么开放协议要解决的核心问题是让“AI Agent”和“产品发现服务”之间拥有稳定的通信语言。协议要定义清楚三件事第一消息格式也就是请求和响应用什么结构表达第二行为语义比如搜索、比较、排序、上下文续传、事件上报各自应该怎样工作第三非功能性要求比如安全认证、错误处理、限流、审计日志。只要 Provider 遵循这套协议AI Agent 就不需要关心底层数据存在哪、用什么语言开发、排序算法是什么。协议还有一个容易被忽略的价值把“产品发现”从单一平台 API 的绑定中解放出来。企业可以把多个自营平台、第三方供应商的产品目录通过同一个网关做接入让 AI Agent 统一调度。对用户而言获得的是更完整的“发现体验”对企业而言增加一个可被 AI Agent 发现的渠道会比单独开发一个 App 或小程序更容易触达用户。正因如此“开放”这两个字很重要协议不应该是某家公司私有标准而应该公开字段定义、允许任何人实现 Provider 或客户端。1.4 协议适用场景我建议不要一上来就追求大而全的协议而是先覆盖高频场景。最核心的接口有三个搜索search、比较compare、事件上报event。搜索解决“从模糊需求到候选集合”比较解决“在候选里做多商品属性对比”事件上报解决“用户最终的选择和反馈回传”。先把这个闭环跑通再逐步扩展推荐、收藏、订阅等能力。本文的完整示例就以这三个场景为核心展开但代码部分会重点实现最常用到的搜索接口另外两个会给出清晰的扩展建议。2. 协议设计参与者、消息模型与核心流程2.1 参与者划分协议里至少包含三类参与者。第一类是 AI Agent也就是调用方它负责理解用户需求、维护对话状态、整合多个 Provider 的返回结果。第二类是 Discovery Provider也就是产品发现能力提供方它可以是电商平台、SaaS 目录、企业内部的商品中心、甚至是第三方数据服务商。第三类是 Identity Provider负责认证和权限管理通常与 Auth Provider 一起工作。在最小实现里Identity 可以通过 API Key 或 OAuth2 Bearer Token 完成不必单独部署服务。AI Agent 和 Discovery Provider 之间是“调用与被调用”的关系但这个关系不能是不受约束的。协议要求每个请求必须有唯一 request_id方便链路追踪每个 Provider 的返回必须携带 search_id便于后续事件上报时关联“第几次搜索”。这两个标识是很多接入方案容易忽略的细节但对线上问题排查至关重要。2.2 消息模型定义为了让不同系统之间通信不产生歧义协议需要用 JSON Schema 或类型定义把消息结构固下来。下面是一份简化但可落地的消息模型。SearchRequest是 AI Agent 发给 Provider 的搜索请求字段包括 request_id、query、context、limit、offset 和一些自定义扩展字段。ProductCard是产品信息的最小载体字段包括 product_id、title、description、price、attributes、source 和 score。attrubutes 里可以放租户自定义属性比如“降噪等级”“续航时长”“发货地区”。{ request_id: req-20250101-001, query: 降噪耳机, context: { user_scene: commute, budget: { currency: CNY, max: 1000 }, preferences: { brands: [demo-a, demo-b] } }, limit: 10, offset: 0, extensions: { sort: score } }响应里除了商品列表还需要返回 total 总数方便 Agent 判断是否要继续翻页同时返回 search_id由 Provider 生成用于关联后续上报事件。这里的 context 可以是半结构化数据但协议会建议标准化几个常见字段user_scene、budget、preferences其他业务字段放入extensions这样既保证通用性又保留扩展空间。{ request_id: req-20250101-001, code: 0, message: ok, search_id: search-xxxx, total: 23, items: [ { product_id: p-001, title: 无线蓝牙降噪耳机 Pro, description: 适合通勤与办公支持主动降噪续航 20 小时。, price: { currency: CNY, amount: 399.00 }, attributes: { category: audio, brand: demo-a, noise_canceling: true }, source: demo-a, score: 0.96 } ] }2.3 核心请求链路一次完整的 AI 中介产品发现通常有五个步骤。第一步用户用自然语言描述需求AI Agent 对需求做意图识别和结构化抽取。第二步AI Agent 根据对话历史构造一个或多个SearchRequest调用一个或多个 Provider。第三步Provider 根据 query 和 context 返回召回结果Agent 对结果做去重、合并、排序。第四步AI Agent 把结果生成推荐文案或商品列表展示给用户。第五步用户做出选择后Agent 调用event接口回传偏好和最终选择帮助 Provider 做后续优化。从协议角度看第二步到第五步都是标准的 HTTP 调用区别只在消息结构。整个链路里最容易出问题的点是第二步到第三步AI Agent 传入的 context 是否真的被 Provider 用来参与召回而不是被忽略。因此协议不只是定义字段还要约定 Provider 应该在 sorting、filtering、ranking 中如何使用这些字段。如果 Provider 不理解某个 context 字段应该明确忽略并在 response 里给出warnings提示而不是让 Agent 误以为自己的约束已经生效。这一点会在后面实战代码里体现。3. 环境准备与最小服务端实现3.1 环境说明为了不把篇幅浪费在某个具体框架的版本上这里用一个非常轻量级的 Python 技术栈做演示。操作系统不限Windows、macOS、Linux 都可以。本地需要安装 Python 3.10 或更高版本然后安装 FastAPI 和 Uvicorn。FastAPI 负责 HTTP API 的定义与参数校验Uvicorn 负责启动本地开发服务器。如果你更习惯 Node.js 或 Go协议本身并不限制语言核心是消息结构和行为语义保持一致代码实现思路可以直接迁移。实战项目的目录结构如下建议直接在本地新建一个aipd-demo文件夹aipd-demo/ ├── app/ │ ├── __init__.py │ ├── schema.py # 协议消息模型 │ ├── store.py # 内存商品数据源 │ └── main.py # FastAPI 服务入口 ├── requirements.txt # 依赖清单 └── client_demo.py # AI Agent 客户端调用示例3.2 安装依赖在aipd-demo目录下创建requirements.txt写入下面几行依赖。版本号建议以你本地实际情况为准如果只需要跑通示例直接安装最新版本通常没有问题。fastapi uvicorn pydantic requests然后执行安装命令pip install -r requirements.txt如果你使用虚拟环境建议先创建并激活虚拟环境再安装依赖。这里不做强制要求但线上开发时强烈建议使用虚拟环境或容器避免依赖污染。3.3 定义协议消息模型在app/schema.py中我们使用 Pydantic 把上一节的 JSON 消息模型变成 Python 类。这样 FastAPI 会自动帮我们完成请求体校验和响应序列化也能在开发阶段获得类型提示。# app/schema.py from typing import List, Optional, Dict, Any from pydantic import BaseModel, Field class Context(BaseModel): AI Agent 传过来的半结构化上下文 user_scene: Optional[str] Field(defaultNone, description用户场景例如 commute、gift) budget: Optional[Dict[str, Any]] Field(defaultNone, description预算例如 {currency: CNY, max: 1000}) preferences: Optional[Dict[str, Any]] Field(defaultNone, description偏好例如品牌、标签) extensions: Optional[Dict[str, Any]] Field(defaultNone, description自定义扩展字段) class SearchRequest(BaseModel): request_id: str Field(..., description调用方生成唯一请求ID) query: str Field(..., description用户查询关键词) context: Optional[Context] Field(defaultNone, description多轮对话上下文) limit: int Field(default10, ge1, le50, description返回条数) offset: int Field(default0, ge0, description分页偏移) extensions: Optional[Dict[str, Any]] Field(defaultNone, description扩展参数) class ProductCard(BaseModel): product_id: str title: str description: str price: Dict[str, Any] Field(..., description价格必须包含 currency 和 amount) attributes: Dict[str, Any] Field(default_factorydict) source: str Field(..., description数据来源标识) score: Optional[float] Field(defaultNone, description相关性分数) class SearchResponse(BaseModel): request_id: str code: int Field(default0, description0 表示成功) message: str Field(defaultok) search_id: str Field(default, descriptionProvider 生成的搜索ID) total: int Field(default0, description满足条件的总数) items: List[ProductCard] Field(default_factorylist, description商品列表) warnings: List[str] Field(default_factorylist, description协议层告警信息)这里有一个细节值得说明price被定义成Dict[str, Any]而不是两个独立字段。原因是在 AI 发现场景里不同供应商的价格结构可能不同有的会包含currency、amount有的还会带original_amount或tax。协议要求至少包含currency和amount额外字段则保留在字典里这样兼容性会更好。3.4 编写内存商品数据源接着在app/store.py中模拟一小部分商品数据。真实项目中这里可以替换成数据库查询、搜索服务调用甚至其他 Provider 的 HTTP 调用。为了让示例专注协议我们使用列表存储。# app/store.py from typing import List, Tuple PRODUCTS [ { id: p-001, title: 无线蓝牙降噪耳机 Pro, description: 适合通勤和办公支持主动降噪续航 20 小时。, price: {currency: CNY, amount: 399.00}, attrs: {category: audio, brand: demo-a, noise_canceling: True}, source: demo-a, }, { id: p-002, title: 头戴式降噪耳机 Max, description: 长续航头戴式耳机适合出差和安静办公。, price: {currency: CNY, amount: 899.00}, attrs: {category: audio, brand: demo-b, noise_canceling: True}, source: demo-b, }, { id: p-003, title: 便携蓝牙音箱 Mini, description: 小体积、防水设计适合户外使用。, price: {currency: CNY, amount: 199.00}, attrs: {category: audio, brand: demo-a, noise_canceling: False}, source: demo-a, }, ] def local_search(query: str, limit: int 10, offset: int 0) - Tuple[List[dict], int]: 基于关键词和描述做一次非常简单的本地搜索 q query.strip().lower() if not q: return [], 0 matched [ product for product in PRODUCTS if q in product[title].lower() or q in product[description].lower() ] return matched[offset:offset limit], len(matched)注意local_search返回两个值当前页数据和匹配总数。分页逻辑很简单但如果接入真实搜索服务排序和过滤应该下沉到存储层而不是在 Python 里硬过滤。3.5 实现 FastAPI 服务在app/main.py中实现协议的服务端。第一步是创建 FastAPI 实例第二步是定义/v1/search接口接收SearchRequest第三步是调用local_search并组装响应。# app/main.py import uuid from fastapi import FastAPI, HTTPException from app.schema import SearchRequest, SearchResponse, ProductCard from app.store import local_search app FastAPI( titleAIPD Provider Demo, version0.1.0, descriptionAIPD (AI-mediated Product Discovery) 协议最小服务端示例, ) app.post(/v1/search, response_modelSearchResponse) async def search(req: SearchRequest): try: matched, total local_search( queryreq.query, limitreq.limit, offsetreq.offset, ) items [ ProductCard( product_idproduct[id], titleproduct[title], descriptionproduct[description], priceproduct[price], attributesproduct[attrs], sourceproduct[source], ) for product in matched ] warnings [] if req.context and req.context.budget: # 示例中不真正实现预算过滤但通过 warning 提示调用方 warnings.append(budget filter is not enforced in demo provider) return SearchResponse( request_idreq.request_id, search_idstr(uuid.uuid4()), totaltotal, itemsitems, warningswarnings, ) except Exception as exc: raise HTTPException(status_code500, detailstr(exc))这段代码里有一个非常实用的协议设计当 Provider 不能处理 context 中的预算条件时不要静默忽略而是通过warnings字段告诉调用方。AI Agent 看到 warning 后可以在文案里主动提示用户“当前搜索没有严格按预算过滤”避免用户产生误解。这个细节会让 AI 产品发现的可信度大幅提升。4. 客户端接入与实战验证4.1 启动服务端进入aipd-demo目录后执行下面的命令启动本地服务uvicorn app.main:app --reload --port 8000启动成功后终端会显示 FastAPI 的访问地址。打开http://localhost:8000/docs你会看到 Swagger 文档页面可以直接在页面上测试/v1/search接口。这个内置文档在开发阶段非常方便也方便前端同学快速理解协议。4.2 使用 curl 快速验证在不写代码的情况下我们可以先用curl调用服务。下面的命令发送一个最简单的搜索请求查询“降噪耳机”。curl -X POST http://localhost:8000/v1/search \ -H Content-Type: application/json \ -d { request_id: req-curl-001, query: 降噪耳机, limit: 5 }预期返回中会看到两条商品记录p-001 和 p-002。因为它们都包含“降噪”和“耳机”关键词。同时返回的search_id是一个随机生成的 UUIDtotal是 2warnings为空。如果你把 query 改成“蓝牙”返回结果会同时包含 p-001 和 p-003因为 p-003 的描述里包含“蓝牙”和“音箱”。4.3 编写 AI Agent 客户端调用在实际的 AI Agent 工程中客户端通常不会直接暴露给用户而是在 Agent 的服务端完成调用。这里编写client_demo.py模拟 AI Agent 内部的联网函数。它负责构造SearchRequest然后解析SearchResponse最后把结果整理成用户可读的文本。# client_demo.py import requests def search_products(base_url: str, query: str, request_id: str): payload { request_id: request_id, query: query, limit: 5, context: { user_scene: commute, budget: {currency: CNY, max: 1000}, }, } resp requests.post(f{base_url}/v1/search, jsonpayload, timeout10) resp.raise_for_status() return resp.json() def format_result(data: dict) - str: lines [] if data.get(warnings): lines.append(注意) for warning in data[warnings]: lines.append(f- {warning}) lines.append(推荐商品) for item in data.get(items, []): price item.get(price, {}) amount price.get(amount, 0) currency price.get(currency, CNY) lines.append(f- {item[title]} ({currency} {amount:.2f})) lines.append(f共找到 {data.get(total, 0)} 件商品) return \n.join(lines) if __name__ __main__: base_url http://localhost:8000 result search_products(base_url, 降噪耳机, req-client-001) print(format_result(result))运行客户端python client_demo.py预期输出仅供参考推荐商品 - 无线蓝牙降噪耳机 Pro (CNY 399.00) - 头戴式降噪耳机 Max (CNY 899.00) 共找到 2 件商品由于我们的 demo Provider 没有真正实现预算过滤所以这里没有输出 warning。如果后面改造成真实数据源并且不支持 context 里的某个字段客户端就会看到 warning。AI Agent 可以根据 warning 决定如何向用户表达这是协议带来的弹性空间。4.4 结果说明与协议行为验证从上面的完整调用可以看出协议的价值在于“两端只依赖同一个消息契约”。AI Agent 端不需要知道 Provider 的数据库结构只需要解析SearchResponseProvider 端不需要关心 Agent 是用了 GPT、Claude 还是本地模型只需要接收SearchRequest。如果要新增一个 Provider只需要按协议实现/v1/search再在 Agent 端的 Provider 列表里登记即可不需要修改 Agent 的消息处理逻辑。反过来如果 Agent 要从“只能搜商品”扩展到“能比较商品”只需要在协议里增加一个/v1/compare端点所有遵守协议的 Provider 都能在同一时间拥有这个能力。5. 安全边界、协议扩展与生产落地5.1 鉴权与最小权限在本地 demo 里接口完全公开任何人都可以调用。生产环境则必须加上身份认证和权限控制。建议使用 API Key 或 OAuth2 Bearer Token 作为调用凭证。每个 AI Agent 只能访问它有权限看到的商品目录并且要对每一次调用做审计。这里需要特别强调最小权限原则AI Agent 的接口令牌不应该拥有修改商品、删除商品、管理用户的权限它只需要aipd:search和aipd:event:write这两类核心权限。在 FastAPI 中可以通过依赖注入轻松实现 API Key 校验。这里给出一个最小示例片段。注意真正生产环境应该使用更安全的密钥存储和轮换机制。# app/auth.py最小鉴权示例 from fastapi import Header, HTTPException API_KEYS { demo-agent: sk-demo-123456, } def verify_api_key(x_api_key: str Header(...)): if x_api_key not in API_KEYS.values(): raise HTTPException(status_code401, detailinvalid api key)然后修改接口定义加入dependencies[Depends(verify_api_key)]。这样客户端调用时就必须携带X-API-Key: sk-demo-123456。关于密钥管理、轮换和最小权限需要结合你实际的 IAM 体系来做协议本身不限定具体实现。5.2 分页、排序与结果稳定性AI 产品发现场景里分页设计比普通 API 要更谨慎。用户在多轮对话中可能多次翻页如果每次翻页时 Provider 内部排序不稳定就会出现重复商品或漏掉商品。因此协议要求 Provider 对同一 query 的排序结果是确定性的。如果排序依赖模型分数建议在响应里返回score并尽量使用tiebreaker字段保证稳定。offset/limit是兼容 REST 习惯的简单方案但数据量很大时性能会变差更推荐游标分页。可以在协议的extensions字段里加入cursor参数用于替代 offset这个扩展不影响核心字段。同时AI Agent 侧也要做去重。多个 Provider 可能返回同一件商品比如同一家店的商品既在自营目录里也在聚合平台里此时source字段和product_id组合是最好的去重键。Agent 在合并结果时可以根据score重新排序也可以保留 Provider 自身的顺序关键是在协议层把这些规则说清楚。5.3 上下文传递的实用策略协议里的context是 AI Agent 与 Provider 之间传递“用户软约束”的主要手段。但注意context 不是无限大的对话历史更不是 Prompt 原文。推荐的做法是AI Agent 在调用 Provider 前先把用户需求做一次结构化抽取只把与商品筛选相关的字段放进 context。例如user_scene、budget、brands、delivery_area这些字段可以标准化至于用户说过的“心情不好”“喜欢简约风”之类的信息如果 Provider 支持语义理解可以放进extensions否则不要强行传递。在 demo 服务里我们特意返回了 budget warning就是为了向 Agent 说明“预算约束没有被执行”。在真实协议里一个更负责的 Provider 应该在服务端实现 budget 过滤而不是把过滤责任推给 Agent。如果暂时做不到就要通过 warning 或显式字段告诉 Agent 哪些 context 字段被忽略了这是协议比普通 REST API 更高级的地方。5.4 版本管理与兼容演进开放协议发布后一定会遇到版本演进。最简单的方式是在 URL 路径中加入版本号例如/v1/search、/v2/search。版本的语义要提前约定好新增字段可以向后兼容修改字段含义或删除字段属于不兼容变更必须升大版本。建议每个字段都使用 optional 方式设计这样新版本可以在旧版本基础上增加信息而不会破坏已有调用方。除了 URL 版本号还建议在响应头中返回协议版本号例如X-AIPD-Version: 1.0。这样 AI Agent 可以快速判断是否调用了旧版本 Provider并做相应适配。如果你把协议发布到公司内部网关还可以结合多环境部署和灰度发布来控制不同版本的上线范围。5.5 限流、超时与降级AI Agent 调用 Provider 的链路比普通前端调用更长一个 Agent 往往要同时调用多个 Provider。为了避免某个慢 Provider 拖垮整体体验客户端必须设置合理的超时时间建议默认 2-3 秒。如果某个 Provider 超时AI Agent 应该自动降级跳过该 Provider不要把错误直接暴露给用户。Provider 端则要提供容量保护和限流避免单个 Agent 的异常重试影响其他调用方。协议里可以增加一个统一的错误响应结构例如error字段包含code、message、details。客户端遇到统一错误结构时可以快速归类错误遇到非标准错误时只能做通用兜底。为了让错误更容易排查每个日志都必须带上request_id和search_id形成完整追踪链。6. 常见问题与排查思路下面整理 AI 中介产品发现协议在落地过程中最常见的几类问题。问题现象常见原因解决思路调用/v1/search返回 401API Key 缺失或错误检查请求头X-API-Key确认令牌有效期和权限范围接口返回 404路径版本号不匹配确认客户端与 Provider 版本一致例如/v1/search响应解析失败Provider 返回结构与协议不一致先用docs接口或 Schema 校验抓包比对字段名和类型商品结果重复多个 Provider 返回同一商品使用source product_id去重再按 score 重新排序返回结果为空但 total 较大分页 offset 计算错误或游标过期检查分页参数推荐使用 cursor 分页context 里的预算没生效Provider 未实现该过滤逻辑检查响应warnings字段联系 Provider 服务方实现请求超时Provider 响应慢或网络链路长客户端缩短超时时间开启降级重试策略多轮对话后结果漂移Provider 排序不稳定要求 Provider 提供稳定排序并返回score日志里查不到问题request_id 没有贯穿全链路通信和日志都统一透传 request_id这些问题的根源大多不是某一个代码 bug而是“没有协议意识”。例如返回结构不一致、版本混乱、上下文语义模糊都是可以先通过协议定义规避的。排错时第一步永远先确认消息结构是否符合协议第二步看有没有warnings或error信息第三步再检查具体业务逻辑。不要一上来就去查数据库或改代码那样效率很低。7. 最佳实践与工程建议7.1 先定契约再写代码在正式开发前先确定一份“最小可用协议”。可以把SearchRequest和SearchResponse的多数字段先定义成 optional然后让两端以这份契约为准进行开发。不要一边写代码一边改字段否则很容易出现 Provider 和 Agent 各维护一套“隐式约定”的情况。建议把协议 Schema 单独做成一个共享包服务端和客户端都从同一个包引用从源头避免字段不一致。7.2 协议的每一层都要可观测AI 中介产品发现链路里用户最终看到的只是一个推荐列表但中间经过了“意图理解 - 调用多Provider - 结果融合 - 文案生成”等多个环节。任何一个环节出问题都会让结果质量下降。因此要在协议层增加可观测性。建议每个 Provider 在响应头里返回X-Request-Id、X-Search-Id、X-Duration-MsAgent 端统一打印到日志。配合链路追踪系统可以让“哪个 Provider 慢”“哪个 Provider 结果差”一目了然。7.3 关注用户授权与隐私边界协议传递 context 时要特别小心隐私信息。用户位置、年龄、收入等敏感信息不能随意传给第三方 Provider必须脱敏或只传化名 ID。每一类 Provider 能拿到哪些字段需要在授权协议里写清楚。AI Agent 在调用 Provider 之前要主动检查当前用户是否已经授权该数据源未授权时跳过调用。最小权限原则同样适用于数据字段不要把所有上下文一股脑传给所有 Provider。7.4 结果要做解释而不仅是给出列表AI 产品发现与普通搜索不同的地方在于用户需要“为什么推荐这个”。协议里的score字段只能表达相关性不足以说明理由。建议在extensions中补充reasons数组例如“符合预算上限”“匹配品牌偏好”“用户评分高”。Provider 可以提供这些理由AI Agent 将它们组合成推荐文案。这个设计会在用户体验上有明显提升也是 AI 产品发现相对传统搜索的增量价值。7.5 让协议的进化沿着真实场景走不要一开始就设计出功能繁多的协议。正确的做法是先跑通“搜索到展示到反馈”的最小闭环然后在真实业务中观察哪些字段缺失、哪些流程无法覆盖。比如对比功能一开始可能不需要但用户经常在几件商品之间犹豫那就应该优先补/v1/compare接口。协议应该以迭代方式演进每次升级都要有日志记录和兼容性说明。8. 总结与下一步这篇文章从实际业务痛点出发完整讨论了“AI 中介产品发现”为什么需要开放协议并围绕SearchRequest、SearchResponse、ProductCard等核心消息模型设计了一套最小可落地的协议。我们用 FastAPI 实现了一个简单的 Provider 服务端并编写了 AI Agent 客户端调用示例同时补充了鉴权、分页、上下文传递、版本演进和可观测性等生产级细节。虽然示例的服务端非常简单但它已经构成了一个完整的协议闭环只要你愿意完全可以照着这套思路把内部商品中心、第三方供应商、甚至其他 Agent 服务都接入进来。下一步建议你先在自己的项目里定义一套最小协议不一定要做得很完整先把“搜索-比较-反馈”这条主线跑通。然后从真实用户反馈里观察协议缺了什么、哪里语义不清楚再按版本机制逐步演进。开放协议不是一次性的架构设计而是持续演进的一套协同规则。如果你正在做 AI Agent 或产品目录相关的项目可以从今天的 /v1/search 开始迈出第一步。