Grok Bot开发实战:从API接入到代购订单自动化
之前在做一个代购类自动化项目时,我卡在了“提示词写好但结果不稳定”和“账户授权流程理不清”这两个环节上。网上关于 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.txt:
fastapi==0.104.1 uvicorn==0.24.0 httpx==0.25.1 pydantic==2.4.2 pydantic-settings==2.0.4 python-dotenv==1.0.0 apscheduler==3.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_KEY=your_grok_api_key GROK_MODEL=grok-2-latest # Link 开放平台配置 LINK_CLIENT_ID=your_link_client_id LINK_CLIENT_SECRET=your_link_client_secret LINK_REDIRECT_URI=http://localhost:8000/auth/callback LINK_AUTH_URL=https://api.link.example.com/oauth/authorize LINK_TOKEN_URL=https://api.link.example.com/oauth/token LINK_ORDER_URL=https://api.link.example.com/v1/orders # 服务配置 APP_HOST=0.0.0.0 APP_PORT=8000注意,以上 URL 需要替换为你实际对接平台的地址。如果你还没有真实环境,可以先写本地 Mock 服务。
4.3 配置管理
文件路径:app/config.py
from 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.py
from 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.py
import 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": f"Bearer {self.api_key}", "Content-Type": "application/json", } async with httpx.AsyncClient(timeout=30) as client: resp = await client.post(self.base_url, json=payload, headers=headers) 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.py
import 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(timeout=15) as client: resp = await client.post(self.token_url, data=payload) 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(timeout=15) as client: resp = await client.post(self.token_url, data=payload) 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": f"Bearer {self.access_token}"} async with httpx.AsyncClient(timeout=30) as client: resp = await client.post(self.order_url, json=order_data, headers=headers) resp.raise_for_status() return resp.json() async def get_order(self, order_id: str) -> dict: """查询订单状态。""" await self._ensure_token() headers = {"Authorization": f"Bearer {self.access_token}"} async with httpx.AsyncClient(timeout=15) as client: resp = await client.get(f"{self.order_url}/{order_id}", headers=headers) resp.raise_for_status() return resp.json()这段代码实现了一个简化版 OAuth2 客户端。真实项目中,token 应该持久化到数据库,而不是保存在内存里。这里为了演示方便,把 token 保存在了实例属性中,实际落地时建议加一层存储抽象。
4.7 订单服务
文件路径:app/order_service.py
from .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_id=user_id, product_url=parse_data.get("product_url"), product_name=parse_data.get("product_name"), spec=parse_data.get("spec"), quantity=parse_data.get("quantity", 1), max_price=parse_data.get("max_price"), note=parse_data.get("note"), ) draft_id = f"DR{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.py
class Notifier: async def send_text(self, user_id: str, message: str): """发送文本通知。 实际项目中可以对接 IM 机器人、邮件或短信服务。 这里只打印日志,方便本地调试。 """ print(f"[NOTIFY] user={user_id}, message={message}")通知服务在生产环境中非常重要。用户确认草稿后,可能等了好几个小时才收到发货通知,如果中间没有任何反馈,体验会很差。建议在以下节点主动通知用户:
- 草稿生成后,请用户确认。
- 订单创建成功后,告知订单号。
- 订单状态变化时,同步最新状态。
- 创建失败时,说明原因和下一步操作。
4.9 主服务入口
文件路径:app/main.py
from 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(title="Grok 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_id="user_001", parse_data=parse_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_id=draft.user_id, product_url=draft.product_url or "", product_name=draft.product_name or "未命名商品", spec=draft.spec, quantity=draft.quantity, max_price=draft.max_price, note=draft.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 返回 401 | API Key 不正确或已过期 | 检查环境变量,重新生成 Key |
| Grok API 返回 429 | 请求频率超出限制 | 增加退避重试,控制并发 |
| Grok 返回内容解析失败 | 模型输出被截断或格式异常 | 增加 max_tokens,优化_extract_json容错逻辑 |
| Link 授权回调报错 | redirect_uri 与平台配置不一致 | 检查回调地址是否完全一致,包括端口 |
| 创建订单返回 401 | access_token 过期 | 刷新 token,或检查 refresh_token 是否有效 |
| 创建订单返回 403 | 权限范围不足 | 确认授权 scope 是否包含order:create |
| 用户确认后又取消 | 流程缺少确认环节 | 在草稿确认前增加二次确认 UI 或指令 |
| 订单状态不更新 | Webhook 地址未配置或回调验签失败 | 检查回调地址、签名算法、重试机制 |
5.2 Grok 输出不稳定的处理
Grok 这类大模型输出天然带有随机性,即使设置temperature=0.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 账户下单成功率,再逐步放开。最后一个提醒:如果你要接入真实代购业务,一定要提前确认业务方是否允许机器人自动下单,以及你的操作是否在合规范围内,避免在未授权场景下使用自动化能力。
