GrokClient类封装API调用:接收api_key初始化会话与认证头,统一错误处理并抛出GrokAPIError;支持/chat/completions非流式与流式请求;自动路由、幂等POST重试及速率限制等待。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜

你需要频繁调用Grok API但每次都要重复写请求头、构造JSON体、处理错误响应,代码散落在各处难以维护和测试。面向对象封装能让你把API调用变成 clean、可复用、带类型提示的类方法调用,而不是一堆requests.post拼凑的脚本。
设计基础Client类
第一步:定义一个GrokClient类,接收api_key作为必填参数,内部自动构建认证头和基础会话。这避免了每次调用都手动拼Authorization字符串,也防止密钥意外泄露到日志或异常堆栈中。
第二步:在__init__中初始化requests.Session(),并设置默认headers={'Content-Type': 'application/json', 'Authorization': f'Bearer {api_key}'}。注意【api_key必须通过构造函数传入,禁止硬编码或从环境变量直接读取后不校验】,否则单元测试无法注入模拟密钥。
第三步:为所有HTTP方法(post、get)封装统一的_error_handler私有方法,当status_code ≥ 400时,尝试解析响应中的code和message字段,抛出带上下文的GrokAPIError异常,而不是裸露的requests.exceptions.HTTPError。
封装/chat/completions接口
方法一:实现chat_completions方法,接受messages(list[dict])、model(str)、temperature(float,默认0.7)三个核心参数。内部将它们组装成标准OpenAI兼容格式的JSON payload,然后调用self._session.post(url, json=payload)。
方法二:支持流式响应。当stream=True时,不使用.json()解析,而是用response.iter_lines()逐行yield每个data: {...}块,并用json.loads(line[6:])提取delta内容。这一步不能用response.json(),否则会阻塞等待完整响应。
注意:流式返回的每条数据不含完整choices字段,只含delta,必须由调用方自行累积content字段——Client类不负责拼接,避免状态耦合。
添加模型路由与自动重试
第一步:在Client初始化时,根据传入的base_url参数决定请求域名。若未指定,则默认指向https://api.x.ai/v1;若传入"https://custom-grok.example.com",则所有请求走该地址。
第二步:为post请求添加指数退避重试逻辑。当遇到503或连接超时,最多重试2次,间隔分别为1秒和2秒。重试前检查是否为幂等操作——只有chat_completions这类POST请求才启用,GET类接口不重试。
第三步:在重试逻辑中,跳过已包含X-RateLimit-Reset头的响应。如果服务端明确返回了重置时间,直接sleep到对应时刻再发起下一次请求,比固定间隔更精准。


















