面向AI商品发现的开放协议PDAP设计实践

最近在搭建 AI 选品与智能推荐链路时有一个问题反复被团队提起AI 代理Agent正在代替用户完成产品发现但现有搜索、推荐、电商开放平台的接口仍然是为“人机交互”设计的。这种错位导致 AI 产品在检索商品时经常需要写大量适配代码而且不同平台返回字段不一致无法形成统一语义。于是我开始整理一种面向 AI 中介产品发现场景的开放协议设计思路。本文就是这份实践笔记包含协议设计动机、消息结构、最小可运行 Demo、常见问题排查和工程建议。适合正在做 AI Agent、智能导购、语义搜索、推荐系统、电商开放平台的同学参考零基础读者也可以先从协议模型部分开始看。1. 什么是 AI 中介的产品发现为什么需要协议1.1 从搜索框到 AI 中介传统产品发现流程中用户会直接在电商平台、搜索引擎或应用内输入关键词然后由系统返回商品列表。整个过程是“人 → 搜索框 → 结果页”。AI 中介产品发现则变成了“人 → AI Agent → 发现服务 → 结果返回给 Agent → Agent 进一步决策”。这里的 AI Agent 可能是聊天机器人、智能采购助手、虚拟导购、企业内部采购系统甚至是另一个 AI 应用。在这种链路中AI Agent 需要完成三件事理解用户模糊意图比如“适合大学生编程练习的轻薄笔记本预算 6000 以内”。把意图转换成结构化的查询参数例如内存、价格区间、品牌偏好。请求产品发现服务并对返回结果进行解释、排序、推荐说明。问题在于如果每个产品发现服务都使用私有接口和自定义字段AI Agent 每接入一个平台就要重新适配。这会导致开发成本高、接口不稳定、语义混乱。1.2 没有协议时会发生什么在没有统一协议的情况下团队通常面临这些问题各平台返回字段不一致。有的返回price有的返回salePrice有的用amount表示价格。查询语义无法统一。有的平台筛选条件只能传分类 ID有的需要传标签数组。缺少通用上下文。AI Agent 的用户偏好、会话信息、约束条件无法在接口间自然传递。无法审计。AI 做了什么决策、推荐了哪些商品、为什么推荐缺少标准化日志结构。扩展困难。接入新的 AI 能力时往往要改服务端接口而不是通过协议扩展解决。本质上产品发现从“机器展示给用户”变成了“机器与机器对话”对话双方需要一个共同的“语言”这就是协议的价值。1.3 一个开放协议要解决的核心问题“面向 AI 中介产品发现的开放协议”重点解决以下问题问题协议设计目标语义不统一定义统一的请求、响应消息结构消除字段歧义上下文丢失将用户意图、约束、会话信息显式建模Agent 可解释性返回推荐理由、分数、候选来源方便 Agent 向用户解释可扩展性通过版本控制和扩展字段兼容未来能力可观测性携带请求 ID、链路 ID支持全链路追踪多 Agent 协作支持 Agent 之间传递发现结果与反馈信号这些设计目标不是单纯为了“做个 API”而是为了让 AI 系统在复杂产品发现场景中具备稳定的协作基础。2. 协议设计目标与整体架构2.1 设计目标协议名称可以沿用项目标题An open protocol for AI-mediated product discovery简称为 PDAPProduct Discovery Agent Protocol下面统一用 PDAP 指代。PDAP 的设计目标可以概括为六条开放性协议定义公开任何一方都可以实现客户端或服务端。中立性不绑定特定电商平台、搜索引擎或大模型供应商。语义清晰请求、响应中的字段有明确业务含义。可扩展允许新增字段和版本演进。可观测支持链路追踪和审计。安全可控支持认证、权限控制和数据最小化。需要说明本文的 PDAP 是一个教学示例协议不是已发布的行业标准。真实环境中设计协议时可以参考这个思路但需要结合具体业务场景做约束、评审和版本管理。2.2 参与者模型PDAP 中涉及四类参与者用户User发起需求的自然人也可以是企业采购系统。AI AgentAgent理解用户意图构造 DiscoveryRequest解析并加工 DiscoveryResponse。发现服务Discovery Service接收请求调用商品库、推荐模型或大模型返回结构化结果。商品数据源Product Catalog提供商品基础数据可能来自电商平台、ERP、供应商系统。四者之间的交互流程如下用户 │ 自然语言表达 ▼ AI Agent │ 1. 构造 DiscoveryRequest ▼ 发现服务 │ 2. 查询商品库 / 调用模型 ▼ 商品数据源 │ 3. 返回候选商品 ▼ 发现服务 │ 4. 构造 DiscoveryResponse ▼ AI Agent │ 5. 自然语言回复 ▼ 用户2.3 核心流程一次完整的产品发现流程可以拆成五个阶段意图理解AI Agent 将用户输入转换成结构化查询条件。请求构建按 PDAP 协议生成 DiscoveryRequest。候选召回发现服务根据条件从商品库召回候选集合。排序解释发现服务通过规则模型或 AI 模型排序并生成推荐理由。响应呈现AI Agent 解析响应结果最终输出给用户。流程的关键点是请求与响应都使用统一协议AI Agent 不直接感知每个商品库的内部实现细节。3. 协议消息定义3.1 请求消息 DiscoveryRequest请求消息的关键是表达“用户想找什么”和“在什么条件下找”。以一个典型请求为例{ protocol_version: 0.1.0, request_id: req_9f8e7d, timestamp: 2025-06-01T10:30:00Z, agent: { agent_id: shopping_assistant_demo, agent_role: shopping_assistant }, user_intent: { query: 适合大学生编程练习的轻薄笔记本, constraints: { max_price: 6000, brands: [Lenovo, HP] }, preferences: { ram_gb_min: 16 } }, context: { region: CN, currency: CNY, session_id: session_xyz, locale: zh-CN } }字段说明protocol_version协议版本号服务端根据它决定解析方式。request_id请求唯一 ID用于追踪和幂等。timestamp请求产生时间使用 ISO 8601 格式。agent发起请求的 Agent 信息便于审计。user_intent.query用户原始意图允许非结构化文本。user_intent.constraints硬性约束条件例如价格上限、品牌白名单。user_intent.preferences软性偏好用于排序加权。context会话上下文包括地区、货币、语言等。这样设计的目的是让 AI Agent 既能传原始自然语言又能传结构化的精确条件兼顾灵活性与可解析性。3.2 响应消息 DiscoveryResponse响应消息需要满足 AI Agent 的展示和解释需求因此不能只返回商品列表还要返回理由。{ protocol_version: 0.1.0, request_id: req_9f8e7d, timestamp: 2025-06-01T10:30:01Z, status: success, products: [ { product_id: p_1001, title: 轻云14 轻薄本 16G512G, score: 0.92, reasons: [内存满足需求, 价格在预算内, 轻薄便携适合移动学习], meta: { price: 5299, currency: CNY, brand: Lenovo } } ], candidates_count: 1, trace_id: trace_abc }响应中几个关键设计点status表示请求处理结果分为success和error。products商品列表按相关性从高到低排列。score关联程度分数范围 0 到 1方便 Agent 做阈值控制。reasons推荐理由数组这是 AI 可解释性的核心Agent 可以直接把它转成自然语言。candidates_count候选商品数量方便 Agent 感知召回范围。trace_id链路追踪 ID便于排查问题。3.3 错误与状态码协议建议使用 HTTP 状态码表达传输层状态同时在响应体中用error对象表达业务错误{ protocol_version: 0.1.0, request_id: req_9f8e7d, timestamp: 2025-06-01T10:30:02Z, status: error, error: { code: UNSUPPORTED_VERSION, message: protocol version 0.0.1 is not supported, details: {} } }建议错误码错误码含义建议处理方式INVALID_JSON请求体不是合法 JSON修复客户端序列化逻辑REQUIRED_FIELD_MISSING缺少必填字段检查协议版本与字段名UNSUPPORTED_VERSION协议版本不支持升级客户端或服务端QUERY_TIMEOUT发现服务处理超时重试或降低复杂度UPSTREAM_ERROR下游商品数据源异常等待恢复避免频繁重试3.4 协议版本约定协议采用主版本.次版本.修订号三段式版本号。主版本变化消息结构不兼容需要强制变更。次版本变化新增可选字段不破坏已有解析。修订号变化纯文档修订或说明调整。服务端在接收到请求时先判断protocol_version是否在支持范围内不在范围内则返回UNSUPPORTED_VERSION。4. 环境准备与工程结构4.1 运行环境为了便于读者快速复现本文示例使用 Python 标准库实现不依赖第三方包。操作系统Windows / macOS / Linux 均可Python 版本3.10 及以上依赖无终端工具任意支持 Python 的终端版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示协议设计思路。4.2 项目目录结构示例工程目录如下pdap-demo/ ├── protocol/ │ └── schema.py # 协议数据结构和校验函数 ├── server.py # 发现服务端 ├── client.py # AI Agent 模拟客户端 └── README.md # 说明文档可选其中protocol/schema.py是核心它定义了协议消息的 Python 数据结构并提供了校验函数。5. 完整代码实现5.1 协议 Schema 定义首先定义协议消息的数据结构和校验逻辑。文件路径pdap-demo/protocol/schema.py PDAP 协议模式定义与校验。 本文件为教学示例字段可根据业务需要扩展。 import re from dataclasses import dataclass, field from typing import Any, Dict, Optional SUPPORTED_PROTOCOL_VERSIONS {0.1.0} REQUEST_ID_PATTERN re.compile(r^[A-Za-z0-9_-]{4,64}$) class ProtocolValidationError(Exception): 协议校验异常。 def __init__(self, code: str, message: str): super().__init__(message) self.code code self.message message def validate_protocol_version(version: str) - None: if version not in SUPPORTED_PROTOCOL_VERSIONS: raise ProtocolValidationError( UNSUPPORTED_VERSION, fprotocol version {version} is not supported, ) def validate_request_id(request_id: Any) - None: if not isinstance(request_id, str) or not REQUEST_ID_PATTERN.match(request_id): raise ProtocolValidationError( REQUIRED_FIELD_MISSING, request_id must be a string of 4-64 chars using [A-Za-z0-9_-], ) def validate_required_fields(payload: Dict[str, Any], required_fields) - None: for field_name in required_fields: if field_name not in payload: raise ProtocolValidationError( REQUIRED_FIELD_MISSING, fmissing required field: {field_name}, ) def parse_discovery_request(raw_body: Dict[str, Any]) - Dict[str, Any]: 将请求 body 解析为 DiscoveryRequest 字典并执行基础校验。 validate_required_fields( raw_body, [protocol_version, request_id, timestamp, agent, user_intent], ) validate_protocol_version(raw_body[protocol_version]) validate_request_id(raw_body[request_id]) if not isinstance(raw_body.get(agent), dict): raise ProtocolValidationError( REQUIRED_FIELD_MISSING, agent must be an object, ) if not isinstance(raw_body.get(user_intent), dict): raise ProtocolValidationError( REQUIRED_FIELD_MISSING, user_intent must be an object, ) return raw_body这里parse_discovery_request对请求做了三层检查必填字段是否存在。协议版本是否支持。字段类型是否合法。这样服务端可以在业务处理之前快速失败避免无效请求进入排序逻辑。5.2 服务端实现文件路径pdap-demo/server.py PDAP 发现服务端演示实现。 使用 Python 标准库 http.server无第三方依赖。 import json import re from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import urlparse from protocol.schema import ( ProtocolValidationError, parse_discovery_request, ) # ---------- 模拟商品库 ---------- PRODUCT_CATALOG [ { product_id: p_1001, title: 轻云14 轻薄本 16G512G, price: 5299, brand: Lenovo, ram_gb: 16, weight_kg: 1.4, tags: [编程, 轻薄, 学习], }, { product_id: p_1002, title: 星锐15 全能本 8G256G, price: 3999, brand: HP, ram_gb: 8, weight_kg: 1.8, tags: [办公, 影音], }, { product_id: p_1003, title: 极客Pro 设计师本 32G1T, price: 8999, brand: Lenovo, ram_gb: 32, weight_kg: 2.1, tags: [编程, 设计, 高性能], }, ] def _safe_float(value, default0.0) - float: try: return float(value) except (TypeError, ValueError): return default def ai_score(product: dict, user_intent: dict) - float: 模拟 AI 排序分数函数。 在实际项目中这里可以替换为 LLM 调用、向量召回、 推荐模型推理或复杂多目标排序服务。 score 0.0 query user_intent.get(query, ) query_terms set(re.findall(r[\w\u4e00-\u9fa5], query)) product_tags set(product.get(tags, [])) title_text product.get(title, ) # 1. 查询词命中商品标签或标题 for term in query_terms: if term in product_tags: score 5.0 if term in title_text: score 3.0 # 2. 约束条件匹配 constraints user_intent.get(constraints, {}) max_price _safe_float(constraints.get(max_price)) if max_price 0 and product.get(price, 0) max_price: score 5.0 brands constraints.get(brands) if isinstance(brands, list) and product.get(brand) in brands: score 2.0 # 3. 偏好条件匹配 preferences user_intent.get(preferences, {}) min_ram _safe_float(preferences.get(ram_gb_min)) if min_ram 0 and product.get(ram_gb, 0) min_ram: score 3.0 # 4. 基础质量分模拟模型置信度 score 0.5 return round(min(score / 10.0, 1.0), 2) def build_reasons(product: dict, user_intent: dict) - list: 根据匹配情况生成可读推荐理由。 reasons [] constraints user_intent.get(constraints, {}) preferences user_intent.get(preferences, {}) query_terms set(re.findall(r[\w\u4e00-\u9fa5], user_intent.get(query, ))) for term in query_terms: if term in product.get(tags, []): reasons.append(f商品标签包含“{term}”) break max_price _safe_float(constraints.get(max_price)) if max_price 0 and product.get(price, 0) max_price: reasons.append(价格在预算范围内) min_ram _safe_float(preferences.get(ram_gb_min)) if min_ram 0 and product.get(ram_gb, 0) min_ram: reasons.append(f内存达到 {min_ram:.0f}GB 以上要求) return reasons def discover_products(user_intent: dict) - list: 候选召回 AI 排序 理由生成。 scored_products [] for product in PRODUCT_CATALOG: score ai_score(product, user_intent) if score 0: continue reasons build_reasons(product, user_intent) scored_products.append( { product_id: product[product_id], title: product[title], score: score, reasons: reasons, meta: { price: product[price], currency: CNY, brand: product[brand], }, } ) scored_products.sort(keylambda x: x[score], reverseTrue) return scored_products def build_success_response(request: dict, products: list, trace_id: str) - dict: return { protocol_version: request[protocol_version], request_id: request[request_id], timestamp: datetime.now(timezone.utc).strftime(%Y-%m-%dT%H:%M:%SZ), status: success, products: products, candidates_count: len(products), trace_id: trace_id, } def build_error_response(request: dict, code: str, message: str) - dict: return { protocol_version: 0.1.0, request_id: request.get(request_id, ), timestamp: datetime.now(timezone.utc).strftime(%Y-%m-%dT%H:%M:%SZ), status: error, error: { code: code, message: message, details: {}, }, } class PDAPHandler(BaseHTTPRequestHandler): protocol_version HTTP/1.1 def _send_json(self, status_code: int, payload: dict): body json.dumps(payload, ensure_asciiFalse).encode(utf-8) self.send_response(status_code) self.send_header(Content-Type, application/json; charsetutf-8) self.send_header(Content-Length, str(len(body))) self.end_headers() self.wfile.write(body) def do_POST(self): parsed_url urlparse(self.path) # 协议端点/v1/discovery if parsed_url.path ! /v1/discovery: self._send_json( 404, { status: error, error: {code: NOT_FOUND, message: endpoint not found}, }, ) return content_length int(self.headers.get(Content-Length, 0)) raw_body self.rfile.read(content_length) # 1. 校验 JSON try: body json.loads(raw_body.decode(utf-8)) except (json.JSONDecodeError, UnicodeDecodeError): self._send_json( 400, { status: error, error: { code: INVALID_JSON, message: request body is not valid JSON, }, }, ) return # 2. 协议字段校验 try: request parse_discovery_request(body) except ProtocolValidationError as exc: self._send_json(400, build_error_response(body, exc.code, exc.message)) return # 3. 业务处理 trace_id ftrace_{request[request_id]} products discover_products(request[user_intent]) response build_success_response(request, products, trace_id) self._send_json(200, response) def log_message(self, format, *args): # 精简终端日志便于阅读 pass def main(): server_address (127.0.0.1, 8000) httpd HTTPServer(server_address, PDAPHandler) print(PDAP discovery server listening on http://127.0.0.1:8000) httpd.serve_forever() if __name__ __main__: main()这段代码里有几个值得注意的地方ai_score是模拟函数示例中采用规则打分。真实项目中建议替换为 LLM 或推荐模型调用。build_reasons生成的推荐理由直接对应响应中的reasons字段。校验逻辑放在业务逻辑之前确保异常请求不会进入排序环节。返回 JSON 时使用ensure_asciiFalse避免中文被转义成\uXXXX。5.3 客户端实现文件路径pdap-demo/client.py PDAP 客户端演示。 模拟 AI Agent 构造 DiscoveryRequest 并发送给发现服务。 import json import time import urllib.request from datetime import datetime, timezone def build_discovery_request() - dict: return { protocol_version: 0.1.0, request_id: freq_{int(time.time() * 1000)}, timestamp: datetime.now(timezone.utc).strftime(%Y-%m-%dT%H:%M:%SZ), agent: { agent_id: shopping_assistant_demo, agent_role: shopping_assistant, }, user_intent: { query: 适合大学生编程练习的轻薄笔记本, constraints: { max_price: 6000, brands: [Lenovo, HP], }, preferences: { ram_gb_min: 16, }, }, context: { region: CN, currency: CNY, session_id: session_xyz, locale: zh-CN, }, } def send_discovery_request( payload: dict, endpoint: str http://127.0.0.1:8000/v1/discovery, api_key: str demo-key, ) - dict: data json.dumps(payload, ensure_asciiFalse).encode(utf-8) request urllib.request.Request( endpoint, datadata, methodPOST, headers{ Content-Type: application/json; charsetutf-8, X-API-Key: api_key, }, ) with urllib.request.urlopen(request, timeout10) as response: return json.loads(response.read().decode(utf-8)) def format_reply(response: dict) - str: 模拟 AI Agent 将协议响应转换成自然语言。 products response.get(products, []) if not products: return 抱歉没有找到符合条件的商品。 lines [为你推荐以下商品] for idx, product in enumerate(products[:3], start1): title product[title] score product[score] reasons .join(product.get(reasons, [])) meta product.get(meta, {}) price meta.get(price, 未知) lines.append( f{idx}. {title}价格{price} 元相关度{score}\n f 推荐理由{reasons} ) lines.append(f\n共召回候选 {response.get(candidates_count, 0)} 件商品。) return \n.join(lines) def main(): payload build_discovery_request() print( 发送 DiscoveryRequest...) print(json.dumps(payload, ensure_asciiFalse, indent2)) response send_discovery_request(payload) print(\n 收到 DiscoveryResponse...) print(json.dumps(response, ensure_asciiFalse, indent2)) print(\n----- AI Agent 格式化输出 -----) print(format_reply(response)) if __name__ __main__: main()客户端代码模拟了 AI Agent 的核心行为构建结构化请求。发送 HTTP 请求。解析协议响应。将响应中的reasons转成自然语言回复。其中format_reply展示了 AI Agent 如何利用协议中的reasons字段实现可解释推荐。5.4 运行与验证在项目根目录执行python server.py服务启动后控制台输出PDAP discovery server listening on http://127.0.0.1:8000再打开另一个终端python client.py预期客户端输出 发送 DiscoveryRequest... { protocol_version: 0.1.0, request_id: req_1710000000123, timestamp: 2025-06-01T10:30:00Z, ... } 收到 DiscoveryResponse... { protocol_version: 0.1.0, request_id: req_1710000000123, timestamp: 2025-06-01T10:30:01Z, status: success, products: [ { product_id: p_1001, title: 轻云14 轻薄本 16G512G, score: 0.92, reasons: [价格在预算范围内, 内存达到 16GB 以上要求], meta: { price: 5299, currency: CNY, brand: Lenovo } } ], candidates_count: 1, trace_id: trace_req_1710000000123 } ----- AI Agent 格式化输出 ----- 为你推荐以下商品 1. 轻云14 轻薄本 16G512G价格5299 元相关度0.92 推荐理由价格在预算范围内内存达到 16GB 以上要求 共召回候选 1 件商品。如果想用 curl 验证也可以直接发送原始 JSONcurl -X POST http://127.0.0.1:8000/v1/discovery \ -H Content-Type: application/json; charsetutf-8 \ -H X-API-Key: demo-key \ -d { protocol_version: 0.1.0, request_id: req_curl_001, timestamp: 2025-06-01T10:30:00Z, agent: {agent_id: curl, agent_role: tester}, user_intent: { query: 轻薄本, constraints: {max_price: 6000}, preferences: {ram_gb_min: 16} }, context: {region: CN, currency: CNY} }5.5 结果说明从运行结果可以看到发现服务成功完成了三个动作校验协议字段未出现INVALID_JSON或REQUIRED_FIELD_MISSING错误。根据用户意图的约束条件召回并排序商品。返回了推荐理由AI Agent 可以直接把这些理由展示给用户。这套最小实现说明了协议的核心价值AI Agent 不需要关心PRODUCT_CATALOG内部如何存储只需要发送标准请求并解析标准响应即可。6. 常见问题与排查思路6.1 常见报错与解决方案问题现象常见原因解决思路服务端返回INVALID_JSON请求体不是合法 JSON或中文编码错误检查客户端序列化确认使用 UTF-8 编码服务端返回REQUIRED_FIELD_MISSING协议字段名拼写不一致或缺少必填字段对照协议定义检查必填字段服务端返回UNSUPPORTED_VERSION客户端协议版本与服务端不兼容升级客户端或服务端协议版本客户端报连接拒绝服务端未启动或端口被占用检查服务进程确认监听端口响应中中文乱码客户端未按 UTF-8 解码使用json.loads(response.read().decode(utf-8))请求超时网络策略限制或服务端处理耗时过长增加超时时间优化服务端逻辑响应中candidates_count为 0约束条件过于严格或商品库中无匹配项放宽条件或检查约束字段类型客户端收到非预期结构导致解析失败响应结构与协议版本不匹配先检查status和error字段再做正常解析在协议对接过程中有一个经验值得单独说明很多“protocol fault”类报错本质不是底层传输坏了而是消息格式与预期不一致。比如数据库客户端报protocol error多数情况是服务端返回报文与客户端解析器预期不符。自研协议同理遇到类似问题先抓取请求和响应原文再对比协议 schema往往比反复调整超时参数更有效。6.2 故障排查清单当协议对接出现问题时建议按以下顺序排查抓取完整请求和响应原文确认 JSON 格式正确。确认protocol_version在服务端支持范围内。对照协议定义检查请求必填字段。确认服务端日志中的异常信息区分协议错误和业务错误。使用 curl 构造最小请求排除客户端代码干扰。检查网络层是否有代理、网关修改请求体。核对数据库或商品库返回是否包含异常空值。7. 最佳实践与工程建议7.1 协议版本管理协议一旦上线就会同时存在多个客户端版本。工程上建议兼容期至少保留两个主版本。新增可选字段时不能改变已有字段语义。废弃字段要提前标记deprecated并给出替代字段。服务端接口路径建议包含主版本号例如/v1/discovery。7.2 安全与身份认证真实环境中发现服务不能像 Demo 一样只依赖X-API-Key的明文头。建议使用 HTTPS 加密传输。生产环境采用 OAuth 2.0 Client Credentials 或 mTLS 做服务间认证。API Key 不写入代码仓库使用环境变量或密钥管理服务。对每个 Agent 设置独立身份标识便于审计和限流。遵守最小权限原则Agent 只能访问自己业务范围内的商品数据。例如客户端可以用环境变量读取 API Keyimport os api_key os.environ.get(PDAP_API_KEY, demo-key)生产部署时避免在代码中硬编码任何密钥。7.3 可观测性与审计AI 推荐链路比传统搜索更容易引发误购、超预算、推荐理由不当等问题。因此需要做到每次请求记录request_id和trace_id。服务端日志打印请求摘要、召回数量、排序分数。对推荐结果做抽样留存用于后续审计。统计响应延迟建立基线超过阈值时告警。对 Agent 调用频率做限流防止异常循环请求打垮下游商品库。日志示例2025-06-01T10:30:01Z INFO request_idreq_xxx trace_idtrace_xxx statussuccess candidates_count1 latency_ms457.4 与真实 AI 模型接入本示例的ai_score是规则函数生产环境可以替换为多种 AI 方案调用大模型接口让模型根据用户意图输出结构化排序理由。使用向量数据库做语义召回再结合规则做硬约束过滤。使用推荐模型产出个性化分数再合并到score字段。多模型集成时保留每个商品分数来源方便复盘。无论采用哪种方案建议保持协议层不变只在服务端内部替换ai_score和build_reasons的实现。这样既能快速迭代 AI 能力又不会破坏已经上线的 Agent 客户端。7.5 数据最小化与隐私AI Agent 在请求中可能会携带用户偏好、位置、会话信息。协议设计上要遵循数据最小化只传输完成发现任务所必需的字段。不将敏感个人信息放入context以外的扩展字段。服务端保存日志时对用户 ID 等字段做脱敏处理。明确数据保留期限超过期限自动清理。跨区域请求时注意合规要求避免把不必要的数据发送到其他区域。8. 总结与下一步本文围绕“An open protocol for AI-mediated product discovery”这条主线完整梳理了一套面向 AI 中介产品发现场景的开放协议设计思路。从协议动机、消息定义、最小实现到工程排错覆盖了 AI Agent 与发现服务之间交互的核心链路。如果你正在做 AI 导购、智能采购、语义搜索或 Agent 平台本文的协议消息结构可以直接作为初版参考如果只是学习协议设计也可以把protocol/schema.py当作一个最小范例来理解“消息即契约”的含义。下一步建议从三个方向继续深入把ai_score替换成真实模型调用观察协议层是否需要增加模型信息字段。增加分页、排序字段和推荐理由模板完善协议表达力。补充服务端限流、熔断、监控能力让协议具备生产环境可用性。建议动手跑一遍示例代码再尝试修改PRODUCT_CATALOG和ai_score观察结果变化。只有亲手改过协议字段并观察过错误日志才能真正理解“开放协议”在 AI 产品发现中的价值。