之前在做一个代购类自动化项目时我卡在了“提示词写好但结果不稳定”和“账户授权流程理不清”这两个环节上。网上关于 Grok Bot 的资料大多是单点功能介绍真正能落地到“关联 Link 账户、处理代购请求”的完整案例非常少。这篇文章就把我整理出来的一套开发思路完整拆开从 Grok Bot 是什么、如何通过 API 接入到 Link 账户授权、代购流程状态机设计再到完整代码和常见报错全部串起来讲一遍。不管是刚接触 Bot 开发的新手还是想在自己业务里接入代购场景的后端开发者都可以跟着这篇文章把链路跑通。1. 背景与核心概念1.1 Grok Bot 是什么Grok Bot 是基于 xAI 旗下 Grok 模型能力构建的对话机器人程序。它本身不是单一产品而是一套可以嵌入到业务系统中的 AI 能力接口。你可以通过官方 API 把 Grok 的对话、文本理解、信息抽取能力接到自己的 Bot 服务里实现自动问答、商品信息解析、订单摘要生成等功能。在代购场景里Grok Bot 的核心价值不是“替用户聊天”而是把用户的非结构化需求转成结构化数据。比如用户发来一段话帮我看下这条链接里的包黑色中号如果价格低于 8000 就下单。如果让开发人员写规则去解析这段话需要维护大量关键词和正则表达式而且用户说话方式稍微变一变就失效。而 Grok Bot 可以根据提示词把这段话转换为 JSON 结构包含商品链接、颜色、尺码、价格阈值、下单条件等信息后续业务代码直接消费这个 JSON 即可。1.2 Link 账户是什么这里的 Link 账户指的是代购业务对接的目标账户体系。不同业务方实现方式不一样有些是开放平台账户有些是电商平台子账号有些甚至是企业内部的订单账户。为了讲清楚原理本文以通用的 OAuth2 授权模型为例来设计对接流程。之所以要关联 Link 账户是因为代购流程中Bot 不只是“回答问题”它需要代替用户去目标平台创建订单、查询物流、获取账单信息。如果没有账户授权Bot 就拿不到订单数据也无法执行下单操作。因此Grok Bot 负责“理解意图”Link 账户负责“执行交易”两者配合才能形成完整闭环。1.3 代购自动化的核心问题代购场景看起来简单实际开发中要解决几个关键问题用户输入不固定有人发链接有人发截图有人只发商品名称。账户授权链路长授权、回调、token 刷新、权限范围都要处理。下单操作不可逆一旦创建订单取消和退款成本很高。状态同步复杂订单从创建、支付、发货到完成每个阶段都要通知用户。所以我们不能只把 Grok Bot 当成“聊天机器人”来对接而是要把整个代购流程当成一个状态机来设计。Grok Bot 只是其中一个模块负责把自然语言转换成结构化指令真正的业务执行还要依赖完整的订单服务和 Link 账户客户端。2. 环境准备与版本说明2.1 运行环境本文示例以 Python 3.10 为例操作系统不限Windows、macOS、Linux 都可以。为了保证可复现性建议使用虚拟环境。python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate2.2 依赖库清单我们需要以下几类库FastAPI搭建 Bot 回调服务和 Webhook 接收端。httpx调用 Grok API 和 Link 账户 API。pydantic定义请求和响应数据模型。python-dotenv管理环境变量。APScheduler可选用于定时轮询订单状态。创建requirements.txtfastapi0.104.1 uvicorn0.24.0 httpx0.25.1 pydantic2.4.2 pydantic-settings2.0.4 python-dotenv1.0.0 apscheduler3.10.4注意版本号只是我本地的测试版本。实际项目请以你安装时的最新稳定版为准不同版本之间的 API 可能有差异。2.3 需要准备的两个账号Grok API Key需要从 xAI 官方渠道申请。没有 API Key 的话可以先使用模拟数据调试流程把代码逻辑跑通后再替换成真实接口。Link 开放平台应用凭证包括 Client ID、Client Secret、回调地址。这个需要根据你对接的平台申请通常在开放平台的“应用管理”页面创建。3. 核心原理与流程设计3.1 整体架构先看整个代购 Bot 的简化流程图我不用复杂图表用文字描述用户消息 ↓ Bot 接收消息 ↓ 调用 Grok API 解析意图 ↓ 生成结构化订单草稿JSON ↓ 推送草稿给用户确认 ↓ 用户确认后调用 Link 账户 API 创建订单 ↓ 轮询订单状态 / 接收 Webhook ↓ 同步状态给用户这个流程中Grok Bot 只是第一阶段。真正决定业务成败的是“用户确认”和“订单状态同步”这两个环节。为了安全起见Bot 永远不应该在未经过用户明确确认的情况下直接下单。3.2 Grok API 调用的设计思路Grok API 的调用方式与主流大模型接口类似采用 HTTP 请求传入消息列表和模型名称返回模型生成的文本。我们可以在提示词中要求模型返回 JSON然后通过代码把 JSON 解析出来。这里有一个关键点不要依赖模型的“裸输出”来执行业务逻辑。模型输出天然不稳定可能返回多余的说明文字、Markdown 代码块标记甚至 JSON 格式错误。所以我们在代码中要做两层处理提示词明确要求只输出 JSON不要输出多余内容。代码在解析时做容错提取 JSON 片段并尝试修复常见格式问题。3.3 Link 账户授权的设计思路Link 账户采用 OAuth2 授权码模式流程如下用户访问授权链接。用户在 Link 平台登录并确认授权。Link 平台回调我们的服务携带授权码。服务用授权码换取 access_token 和 refresh_token。后续请求携带 access_token。access_token 过期后用 refresh_token 刷新。这里的核心坑点是 token 过期。很多开发者第一次对接时只保存了 access_token结果第二天调用接口就报 401。设计时必须把 refresh_token 一起保存并实现自动刷新机制。3.4 代购流程状态机代购订单不能只有“创建”和“完成”两个状态我建议按下面这样设计状态含义触发条件DRAFT草稿待用户确认Grok 解析完成生成结构化订单草稿CONFIRMED用户已确认用户点击确认或回复确认指令CREATING正在创建订单调用 Link 账户 API 下单中CREATED订单已创建Link 平台返回订单号PAID已支付支付回调触发SHIPPED已发货物流状态更新COMPLETED已完成用户确认收货FAILED失败创建订单异常CANCELLED已取消用户取消或超时未确认这样设计的好处是每个状态都有明确的触发条件和对应处理逻辑Bot 不会因为中间环节出错而丢失上下文。4. 完整实战案例4.1 创建项目结构我们先创建如下项目结构grok-link-bot/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── config.py │ ├── models.py │ ├── grok_client.py │ ├── link_client.py │ ├── order_service.py │ └── notifier.py ├── .env.example ├── requirements.txt └── README.md4.2 环境变量配置创建.env.example文件# Grok API 配置 GROK_API_KEYyour_grok_api_key GROK_MODELgrok-2-latest # Link 开放平台配置 LINK_CLIENT_IDyour_link_client_id LINK_CLIENT_SECRETyour_link_client_secret LINK_REDIRECT_URIhttp://localhost:8000/auth/callback LINK_AUTH_URLhttps://api.link.example.com/oauth/authorize LINK_TOKEN_URLhttps://api.link.example.com/oauth/token LINK_ORDER_URLhttps://api.link.example.com/v1/orders # 服务配置 APP_HOST0.0.0.0 APP_PORT8000注意以上 URL 需要替换为你实际对接平台的地址。如果你还没有真实环境可以先写本地 Mock 服务。4.3 配置管理文件路径app/config.pyfrom pydantic_settings import BaseSettings class Settings(BaseSettings): grok_api_key: str grok_model: str link_client_id: str link_client_secret: str link_redirect_uri: str link_auth_url: str link_token_url: str link_order_url: str app_host: str 0.0.0.0 app_port: int 8000 class Config: env_file .env env_file_encoding utf-8 settings Settings()使用 pydantic-settings 的好处是可以自动从.env文件读取配置并且字段类型有保障。如果有人忘了配置某个环境变量启动时会直接报错避免到运行时才发现问题。4.4 数据模型定义文件路径app/models.pyfrom pydantic import BaseModel from enum import Enum from typing import Optional class OrderStatus(str, Enum): DRAFT DRAFT CONFIRMED CONFIRMED CREATING CREATING CREATED CREATED PAID PAID SHIPPED SHIPPED COMPLETED COMPLETED FAILED FAILED CANCELLED CANCELLED class GrokParseRequest(BaseModel): user_message: str class GrokParseResponse(BaseModel): product_url: Optional[str] None product_name: Optional[str] None spec: Optional[str] None quantity: int 1 max_price: Optional[float] None note: Optional[str] None need_confirm: bool True class OrderDraft(BaseModel): user_id: str product_url: Optional[str] None product_name: Optional[str] None spec: Optional[str] None quantity: int 1 max_price: Optional[float] None note: Optional[str] None status: OrderStatus OrderStatus.DRAFT class OrderCreateRequest(BaseModel): user_id: str product_url: str product_name: str spec: Optional[str] None quantity: int 1 max_price: Optional[float] None note: Optional[str] None数据模型是整个服务的地基。这里我把草稿订单和创建订单的参数分开目的是确保“用户确认前”和“用户确认后”的数据结构分离避免未确认的数据被直接提交到 Link 平台。4.5 Grok 客户端文件路径app/grok_client.pyimport httpx import json import re from .config import settings class GrokClient: def __init__(self): self.api_key settings.grok_api_key self.model settings.grok_model # 实际接口地址以官方文档为准这里只是示例 self.base_url https://api.x.ai/v1/chat/completions def _extract_json(self, text: str) - dict: 从模型输出中提取 JSON并做基础修复。 # 尝试直接解析 text text.strip() if text.startswith(): text re.sub(r^(?:json)?\s*, , text) text re.sub(r\s*$, , text) try: return json.loads(text) except json.JSONDecodeError: pass # 尝试提取第一个 { 到最后一个 } 之间的内容 start text.find({) end text.rfind(}) if start ! -1 and end ! -1 and end start: candidate text[start:end 1] try: return json.loads(candidate) except json.JSONDecodeError: raise ValueError(Grok 返回内容无法解析为 JSON) raise ValueError(Grok 返回内容中未找到 JSON) async def parse_purchase_intent(self, user_message: str) - dict: 把用户消息解析为结构化订单草稿。 system_prompt 你是一个代购订单信息提取助手。请从用户消息中提取以下字段并严格输出 JSON - product_url: 商品链接没有则为 null - product_name: 商品名称没有则为 null - spec: 规格信息例如颜色、尺码、型号没有则为 null - quantity: 数量默认 1 - max_price: 最高可接受价格没有则为 null - note: 用户其他备注没有则为 null - need_confirm: 是否需要用户确认默认 true 要求 1. 只输出 JSON不要输出任何解释文字。 2. 如果信息不足字段设为 null不要猜测。 3. 数值字段使用数字类型不要用字符串。 payload { model: self.model, messages: [ {role: system, content: system_prompt}, {role: user, content: user_message}, ], temperature: 0.1, response_format: {type: json_object}, } headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, } async with httpx.AsyncClient(timeout30) as client: resp await client.post(self.base_url, jsonpayload, headersheaders) resp.raise_for_status() data resp.json() content data[choices][0][message][content] return self._extract_json(content)这里有几个细节值得说明temperature设置为 0.1减少模型输出的随机性让结构化解析更稳定。response_format要求模型返回 JSON 对象。不是所有模型都支持这个参数如果不支持需要去掉然后在代码里做容错。_extract_json方法处理了模型可能输出代码块标记或前后多余文字的问题。这在真实项目中非常有用因为大模型输出并不是 100% 可控的。4.6 Link 账户客户端文件路径app/link_client.pyimport time import httpx from .config import settings class LinkClient: def __init__(self): self.client_id settings.link_client_id self.client_secret settings.link_client_secret self.redirect_uri settings.link_redirect_uri self.auth_url settings.link_auth_url self.token_url settings.link_token_url self.order_url settings.link_order_url self.access_token None self.refresh_token None self.expires_at 0 def build_authorize_url(self, state: str) - str: 生成用户授权链接。 params { response_type: code, client_id: self.client_id, redirect_uri: self.redirect_uri, state: state, scope: order:create order:read, } return f{self.auth_url}?{httpx.QueryParams(params)} async def exchange_code(self, code: str): 用授权码换取 token。 payload { grant_type: authorization_code, code: code, redirect_uri: self.redirect_uri, client_id: self.client_id, client_secret: self.client_secret, } async with httpx.AsyncClient(timeout15) as client: resp await client.post(self.token_url, datapayload) resp.raise_for_status() data resp.json() self.access_token data[access_token] self.refresh_token data.get(refresh_token, ) self.expires_at time.time() int(data.get(expires_in, 7200)) async def refresh_access_token(self): 刷新 access_token。 payload { grant_type: refresh_token, refresh_token: self.refresh_token, client_id: self.client_id, client_secret: self.client_secret, } async with httpx.AsyncClient(timeout15) as client: resp await client.post(self.token_url, datapayload) resp.raise_for_status() data resp.json() self.access_token data[access_token] self.refresh_token data.get(refresh_token, self.refresh_token) self.expires_at time.time() int(data.get(expires_in, 7200)) async def _ensure_token(self): 确保 access_token 有效。 if not self.access_token: raise RuntimeError(Link 账户未授权请先引导用户完成授权) if time.time() self.expires_at - 60: if not self.refresh_token: raise RuntimeError(refresh_token 不存在需要重新授权) await self.refresh_access_token() async def create_order(self, order_data: dict) - dict: 创建订单。 await self._ensure_token() headers {Authorization: fBearer {self.access_token}} async with httpx.AsyncClient(timeout30) as client: resp await client.post(self.order_url, jsonorder_data, headersheaders) resp.raise_for_status() return resp.json() async def get_order(self, order_id: str) - dict: 查询订单状态。 await self._ensure_token() headers {Authorization: fBearer {self.access_token}} async with httpx.AsyncClient(timeout15) as client: resp await client.get(f{self.order_url}/{order_id}, headersheaders) resp.raise_for_status() return resp.json()这段代码实现了一个简化版 OAuth2 客户端。真实项目中token 应该持久化到数据库而不是保存在内存里。这里为了演示方便把 token 保存在了实例属性中实际落地时建议加一层存储抽象。4.7 订单服务文件路径app/order_service.pyfrom .models import OrderDraft, OrderStatus, OrderCreateRequest class OrderService: def __init__(self): # 实际项目中请替换为数据库存储 self._orders {} self._counter 1 def create_draft(self, user_id: str, parse_data: dict) - OrderDraft: draft OrderDraft( user_iduser_id, product_urlparse_data.get(product_url), product_nameparse_data.get(product_name), specparse_data.get(spec), quantityparse_data.get(quantity, 1), max_priceparse_data.get(max_price), noteparse_data.get(note), ) draft_id fDR{self._counter:06d} self._counter 1 self._orders[draft_id] draft return draft_id, draft def confirm_draft(self, draft_id: str) - OrderDraft: draft self._orders.get(draft_id) if not draft: raise ValueError(草稿不存在) if draft.status ! OrderStatus.DRAFT: raise ValueError(订单状态不是草稿无法确认) draft.status OrderStatus.CONFIRMED return draft def update_status(self, draft_id: str, status: OrderStatus): draft self._orders.get(draft_id) if not draft: raise ValueError(订单不存在) draft.status status return draft def get_order(self, draft_id: str) - OrderDraft: return self._orders.get(draft_id)这里的订单存储使用字典是为了让示例代码跑起来足够简单。生产环境建议使用 Redis 或数据库存储并且要加分布式锁避免多个 Webhook 回调同时修改同一个订单状态。4.8 通知服务文件路径app/notifier.pyclass Notifier: async def send_text(self, user_id: str, message: str): 发送文本通知。 实际项目中可以对接 IM 机器人、邮件或短信服务。 这里只打印日志方便本地调试。 print(f[NOTIFY] user{user_id}, message{message})通知服务在生产环境中非常重要。用户确认草稿后可能等了好几个小时才收到发货通知如果中间没有任何反馈体验会很差。建议在以下节点主动通知用户草稿生成后请用户确认。订单创建成功后告知订单号。订单状态变化时同步最新状态。创建失败时说明原因和下一步操作。4.9 主服务入口文件路径app/main.pyfrom fastapi import FastAPI, Request, Query from fastapi.responses import RedirectResponse from .config import settings from .models import GrokParseRequest from .grok_client import GrokClient from .link_client import LinkClient from .order_service import OrderService from .notifier import Notifier app FastAPI(titleGrok Link Bot) grok_client GrokClient() link_client LinkClient() order_service OrderService() notifier Notifier() app.get(/) async def index(): return {message: Grok Link Bot is running} app.get(/auth/login) async def auth_login(): 引导用户跳转到 Link 平台授权。 state random_state_string url link_client.build_authorize_url(state) return RedirectResponse(url) app.get(/auth/callback) async def auth_callback(code: str Query(...), state: str Query(...)): Link 平台回调地址用授权码换取 token。 await link_client.exchange_code(code) return {message: 授权成功可以开始使用 Bot 代购服务} app.post(/bot/parse) async def parse_message(req: GrokParseRequest): 接收用户消息调用 Grok 解析意图。 parse_result await grok_client.parse_purchase_intent(req.user_message) # 创建草稿订单 draft_id, draft order_service.create_draft( user_iduser_001, parse_dataparse_result, ) # 通知用户确认 await notifier.send_text( draft.user_id, f订单草稿已生成{draft_id}\n f商品{draft.product_name}\n f规格{draft.spec}\n f数量{draft.quantity}\n f最高价{draft.max_price}\n f请回复确认后下单。, ) return { draft_id: draft_id, draft: draft.model_dump(), need_confirm: parse_result.get(need_confirm, True), } app.post(/bot/confirm/{draft_id}) async def confirm_order(draft_id: str): 用户确认草稿后创建真实订单。 draft order_service.confirm_draft(draft_id) # 构建 Link 平台需要的订单数据 order_data OrderCreateRequest( user_iddraft.user_id, product_urldraft.product_url or , product_namedraft.product_name or 未命名商品, specdraft.spec, quantitydraft.quantity, max_pricedraft.max_price, notedraft.note, ) try: result await link_client.create_order(order_data.model_dump()) order_service.update_status(draft_id, OrderStatus.CREATED) await notifier.send_text(draft.user_id, f订单创建成功订单号{result.get(order_id)}) return {success: True, order_id: result.get(order_id)} except Exception as e: order_service.update_status(draft_id, OrderStatus.FAILED) await notifier.send_text(draft.user_id, f订单创建失败{str(e)}) return {success: False, error: str(e)} app.get(/order/{draft_id}) async def get_order(draft_id: str): 查询订单详情。 draft order_service.get_order(draft_id) if not draft: return {error: 订单不存在} return draft.model_dump()4.10 启动服务创建.env文件并填入真实配置后运行uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload启动成功后访问http://localhost:8000/docs可以看到 FastAPI 自动生成的 API 文档可以直接在 Swagger UI 里测试/bot/parse和/bot/confirm/{draft_id}接口。测试时如果还没有真实的 Grok API Key可以在parse_purchase_intent里先写死返回数据例如async def parse_purchase_intent(self, user_message: str) - dict: # 临时 Mock便于联调 return { product_url: https://example.com/item/123, product_name: 示例商品, spec: 黑色 中号, quantity: 1, max_price: 8000, note: None, need_confirm: True, }这样能先把整个下单链路跑通等拿到真实 Key 后再替换。5. 常见问题与排查思路5.1 常见问题排查表问题现象常见原因解决思路Grok API 返回 401API Key 不正确或已过期检查环境变量重新生成 KeyGrok API 返回 429请求频率超出限制增加退避重试控制并发Grok 返回内容解析失败模型输出被截断或格式异常增加 max_tokens优化_extract_json容错逻辑Link 授权回调报错redirect_uri 与平台配置不一致检查回调地址是否完全一致包括端口创建订单返回 401access_token 过期刷新 token或检查 refresh_token 是否有效创建订单返回 403权限范围不足确认授权 scope 是否包含order:create用户确认后又取消流程缺少确认环节在草稿确认前增加二次确认 UI 或指令订单状态不更新Webhook 地址未配置或回调验签失败检查回调地址、签名算法、重试机制5.2 Grok 输出不稳定的处理Grok 这类大模型输出天然带有随机性即使设置temperature0.1也无法保证每次都输出纯 JSON。我的建议是在提示词中明确“只输出 JSON”。代码里分层容错先尝试完整解析再尝试截取 JSON 片段。解析失败时记录原始输出方便复盘优化提示词。生产环境可以使用 Pydantic 的 validator 对字段做二次校验类型不对就拒绝。5.3 token 刷新失败token 刷新失败通常有两个原因refresh_token 过期。有些平台 refresh_token 有效期很短过期后只能重新走授权流程。refresh_token 被平台撤销。用户在 Link 平台手动取消了第三方应用授权。解决方案是做好异常捕获当刷新失败时主动通知用户重新授权而不是默默重试导致死循环。5.4 下单成功但通知丢失如果使用异步任务创建订单可能出现“订单创建成功但回调通知失败”的情况。建议把订单创建结果写入数据库。通知失败时做定时重试。提供主动查询接口用户可以通过订单号随时查询状态。6. 最佳实践与工程建议6.1 安全边界代购 Bot 涉及真实交易安全是第一位。以下几点必须注意API Key 和 Client Secret 绝不能写进代码仓库要放到环境变量或密钥管理服务中。所有令牌统一走服务端保存前端永远拿不到 access_token。日志中要对 token、用户手机号、地址等敏感信息脱敏。下单接口要做幂等控制防止用户重复点击导致重复下单。6.2 配置管理本文示例把配置写在了.env文件里这适合本地开发。生产环境建议使用配置中心管理多环境配置。按环境拆分dev、test、prod。敏感配置加密存储。配置变更走审批和灰度流程。6.3 异常处理与重试调用 Grok API 和 Link API 都属于外部依赖必须做好超时控制和重试。建议网络超时设置合理值Grok 解析可以给 30 秒。对 429 和 5xx 错误做指数退避重试。重试次数限制在 3 次以内避免雪崩。下单接口不建议无脑重试因为重复提交可能创建多个订单。应该先查询订单状态再决定是否重试。6.4 订单状态一致性代购流程涉及多个外部系统订单状态可能出现不一致。比如 Link 平台回调说订单已支付但我们本地数据库还是 CREATED。这时建议以 Link 平台的状态为准定期拉取订单状态做对账。状态变更用乐观锁避免并发覆盖。状态变化记录审计日志方便追溯。6.5 提示词工程优化Grok Bot 在代购场景中的效果很大程度上取决于提示词质量。我的经验是提示词里提供具体字段列表和类型要求。用少量示例说明输入输出对应关系。温度调低减少随机性。解析结果加校验不合法的数据直接拒绝。如果你发现 Grok 解析经常出错可以先人工整理 20 条真实用户消息跑一遍提示词把错误结果收集起来针对性地修改提示词。这个迭代过程对效果提升非常明显。7. 总结与学习路线这篇文章围绕 Grok Bot 关联 Link 账户代购这个场景完整梳理了从 API 接入、账户授权、订单状态机设计到代码实现的整个链路。你可以先按照示例代码把本地服务跑起来用 Mock 数据验证整条代购流程等确认没有问题后再申请真实 API Key、对接真实 Link 平台。如果继续深入建议按下面的路线学习先掌握 FastAPI 的基础用法熟悉依赖注入和异步处理。再深入学习 OAuth2 授权码模式理解 token 刷新和权限范围管理。然后研究数据库设计把内存版订单存储换成 MySQL 或 PostgreSQL。接着做消息队列把下单、回调、通知等操作异步化。最后关注监控告警把 Grok 调用耗时、下单成功率、token 刷新失败率等指标都监控起来。代购类 Bot 的难点不在“调通 API”而在“稳定地处理真实业务”。尤其是订单创建这个环节一次重复下单就可能导致损失。实际项目中建议先跑通小流量灰度验证 Grok 解析准确率和 Link 账户下单成功率再逐步放开。最后一个提醒如果你要接入真实代购业务一定要提前确认业务方是否允许机器人自动下单以及你的操作是否在合规范围内避免在未授权场景下使用自动化能力。