【Python LLM小说自动化生成实战】5｜LLM客户端封装

目录一、依赖安装openai 库部署二、使用 Cline 生成基础代码三、代码优化四、模块核心设计思路讲解4.1 核心方法4.2 工程细节五、项目联动逻辑回顾收尾 下集预告上一节我们完成了提示词模块prompt.py的编写把所有系统指令统一管理好了。但现在还差关键一环怎么把提示词、用户请求发送给 DeepSeek 大模型并且稳妥接收返回结果我们单独新建llm_client.py做成项目唯一对接大模型的入口所有模块想要调用 AI全部通过这个文件。集中处理重试、异常、文本清洗、JSON 解析一处封装全局复用。大模型客户端说白了就是通过SDK发送请求到大模型的运营商然后使用人家的这个模型。一、依赖安装openai 库部署我们使用openai第三方 SDK通过它我们可以发送请求到不同的运营商去使用人家的模型。它除了 OpenAI 官方还兼容 DeepSeek、智谱等绝大多数兼容 OpenAI 接口格式的大模型服务商。 终端执行安装指令uv add openai不过可能会遇到下面这段报错这是因为该指令默认去国外官方地址files.pythonhosted.org下载导致网络超时属于正常现象。解决方案临时使用阿里云国内镜像单次生效uv add openai --index-url https://mirrors.aliyun.com/pypi/simple如果你不想每次安装包都额外追加镜像地址可以设置全局镜像永久生效uv config set index-url https://mirrors.aliyun.com/pypi/simple配置完成后后续直接执行uv add xxx自动走阿里云国内源。二、使用 Cline 生成基础代码我们打开 Cline 插件输入指令/src/llm_client.py 创建LLM调用模块--这是全项目唯一和模型打交道的地方其他代码都通过它调用LLM。需要两种调用方式● 普通对话发消息、拿回文本。要对网络错误/限流做自动重试退避一下再试别一失败就崩。● 要 JSON 的对话发消息、拿回一个解析好的字典。这里有个现实问题—— LLM 经常把 JSON 包在代码块里或夹带废话所以拿回来要先清洗再解析 如果还是解析失败就把上次的错误输出回喂给模型让它自己改一次再给。● 定义清晰的异常区分调用本身失败和JSON 解析失败。 真正的 openai 调用要延迟导入这样没装 openai 的时候纯逻辑也能单独测试。输入指令后切换到Plan查看 AI 规划的实现思路核对逻辑是否符合需求规划确认无误切换Act自动生成代码通读生成源码确认逻辑完整后保存文件。不熟悉 Cline 操作的朋友可以回看本专栏前面环境搭建章节。三、代码优化精简函数参数小编在审查代码的时候观察发现初次生成的代码里chat、chat_json两个方法存在大量冗余参数不利于后续维护。如果有和我类似问题的小伙伴可以继续给 Cline 发送优化指令简化参数设计chat与chat_json函数没必要这么多参数按以下方式编写即可。参数prompt: 用户消息system: 可选的系统提示词角色设定、输出约束等temperature: 采样温度越高越随机优化完成后两个对外调用方法接口更加简洁后续业务层调用的时候可读性更高。四、模块核心设计思路讲解4.1 两套对外核心方法chat () 普通对话接口单纯发送指令获取 AI 返回的原始文本。自带重试机制遇到网络波动、接口限流不会直接崩溃等待一段时间自动重试。适合章节正文扩写场景。chat_json () 结构化 JSON 接口专门用于生成小说大纲。 这里必须重点说一个实战痛点大模型很喜欢自作主张把 JSON 包裹在json代码块内或者前后附带一堆解释文字直接解析会抛出异常。 所以这套函数内置处理流程 ① 清洗返回文本剔除 markdown 标记、多余说明文字 ② 尝试解析字典 ③ 解析失败时把错误内容重新传给 AI要求模型修正格式二次尝试。4.2 两个很关键的工程细节自定义异常分类区分「网络请求失败」和「JSON 解析失败」两类错误上层代码可以针对性捕获、处理不同问题方便调试定位 bug。openai 延迟导入不在文件头部全局导入 openai 库。只有真正发起调用时才加载模块。好处没有安装依赖的环境下其他纯数据逻辑代码也能正常运行方便单元测试。文末附带我调试完成后的成品代码大家可以用来对比自己 Cline 生成的内容查漏补缺。五、项目联动逻辑回顾llm_client.py读取.env 里的 API_KEY、BASE_URL、模型名称、最大重试次数等配置业务代码调用本模块的 chat/chat_json不需要关心底层接口请求细节搭配 prompt.py 内的系统提示词组合成完整请求发送给 DeepSeek获取返回内容清洗、校验后向上层返回结果交给 Pydantic Schema 完成数据校验。收尾 下集预告我们搭建好了整个项目的通信中枢所有和大模型交互的逻辑全部收拢在这一个文件内重试、容错、格式清洗全部内置。下一节预告【Python LLM 小说自动化生成实战】6导出模块开发实现小说内容一键导出 Markdown、Word 文档 本系列专栏合集传送门小白入门从零实战开发Python LLM小说自动化生成系统 LLM 调用模块 —— 全项目唯一和模型打交道的地方。 异常体系 -------- LLMError 基类 ├── LLMAPIError API 调用失败网络、限流、超时等重试耗尽后抛出 └── LLMJSONError JSON 解析失败即使经过自修正后仍失败 核心函数 -------- chat(messages, ...) - str 普通对话返回文本 chat_json(messages, ...) - dict 要 JSON 的对话返回解析好的 dict 注意openai 库采用延迟导入只有首次实际调用 LLM 时才会 import openai。 这样没装 openai 的环境也能安全导入本模块比如纯逻辑测试。 from __future__ import annotations import json import re import time from typing import Any from LLM.src import config # --------------------------------------------------------------------------- # 异常定义 # --------------------------------------------------------------------------- class LLMError(Exception): LLM 调用相关异常的基类。 class LLMAPIError(LLMError): API 调用失败网络、限流、超时、鉴权等。 class LLMJSONError(LLMError): LLM 输出的 JSON 解析失败即使经过自修正后。 def __init__( self, message: str, raw_response: str | None None, corrected_response: str | None None, ) - None: self.raw_response raw_response self.corrected_response corrected_response super().__init__(message) # --------------------------------------------------------------------------- # 内部工具 # --------------------------------------------------------------------------- _client: Any | None None def _get_client() - Any: 延迟导入 openai 并返回 OpenAI 客户端实例单例。 global _client if _client is None: import openai # noqa: F401 — 延迟导入纯逻辑测试时可跳过 _client openai.OpenAI(api_keyconfig.api_key, base_urlconfig.base_url) return _client def _call_openai( messages: list[dict[str, str]], *, model: str, temperature: float, max_tokens: int, max_retries: int, ) - str: 调用 LLM 并返回文本。 内部处理指数退避重试2^attempt 秒覆盖网络断开、限流、超时等。 所有重试耗尽后抛出 LLMAPIError。 client _get_client() last_exception: Exception | None None for attempt in range(1, max_retries 1): try: response client.chat.completions.create( modelmodel, messagesmessages, temperaturetemperature, max_tokensmax_tokens, ) content: str | None response.choices[0].message.content return content or except Exception as e: last_exception e if attempt max_retries: sleep_seconds 2**attempt time.sleep(sleep_seconds) raise LLMAPIError( fAPI 调用失败已重试 {max_retries} 次, ) from last_exception def _clean_json(text: str) - str: 从 LLM 输出的文本中提取最有可能的 JSON 字符串。 处理步骤按优先级 1. 去除首尾空白 2. 尝试提取 markdown 代码块 json ... 3. 尝试提取 markdown 代码块 ... 4. 以 { 开头则直接返回 5. 正则提取第一个 { ... } 块 6. 全部失败则原样返回交由 json.loads 报错 text text.strip() # 提取 json ... m re.search(r(?:json)\s*\n?(.*?), text, re.DOTALL) if m: candidate m.group(1).strip() if candidate: return candidate # 提取 ... m re.search(r\s*\n?(.*?), text, re.DOTALL) if m: candidate m.group(1).strip() if candidate: return candidate # 直接以 { 开头 if text.startswith({): return text # 正则取第一个 { ... } m re.search(r(\{.*\}), text, re.DOTALL) if m: return m.group(1).strip() return text def _try_parse_json(raw: str) - dict[str, Any]: 尝试将 raw 文本清洗后解析成 dict。 返回 (parsed_dict, error_message) 元组。 成功时 error_message 为 None失败时 parsed_dict 为 None。 cleaned _clean_json(raw) try: return json.loads(cleaned) except json.JSONDecodeError as e: raise LLMJSONError( fJSON 解析失败{e}\n清洗后文本{cleaned[:500]}, raw_responseraw, ) from e # --------------------------------------------------------------------------- # 公开 API # --------------------------------------------------------------------------- def chat( messages: list[dict[str, str]], *, model: str | None None, temperature: float | None None, max_tokens: int | None None, max_retries: int | None None, ) - str: 普通对话发消息、拿回文本。 遇到网络错误 / 限流会自动以指数退避重试。 所有参数均有合理的默认值来自 config也可按需覆盖。 return _call_openai( messages, modelmodel or config.model, temperaturetemperature if temperature is not None else config.chaper_temperature, max_tokensmax_tokens or config.max_tokens, max_retriesmax_retries or config.max_retries, ) def chat_json( messages: list[dict[str, str]], *, model: str | None None, temperature: float | None None, max_tokens: int | None None, max_retries: int | None None, ) - dict[str, Any]: 要 JSON 的对话发消息、拿回一个解析好的字典。 流程 1. 调用 chat() 获取原始响应 2. 清洗、解析 JSON 3. 如果解析失败把错误回喂给模型让模型自修正一次 4. 仍失败则抛 LLMJSONError携带原始响应和修正响应 自修正时会在原 messages 末尾追加一条消息 说明「你刚才的输出不是合法 JSON请只输出 JSON」。 raw chat( messages, modelmodel, temperaturetemperature, max_tokensmax_tokens, max_retriesmax_retries, ) try: return _try_parse_json(raw) except LLMJSONError as first_err: pass # ----- 自修正回喂错误信息让模型自己改一次 ----- correction_prompt ( 你上面的输出不是合法的 JSON。请只输出合法的 JSON 对象 不要添加任何 markdown 格式标记不要用 不要添加任何额外文字。\n\n f你的输出\n{raw}\n\n f解析错误\n{first_err} ) corrected_messages [ *messages, {role: assistant, content: raw}, {role: user, content: correction_prompt}, ] raw2 chat( corrected_messages, modelmodel, temperaturetemperature, max_tokensmax_tokens, max_retriesmax_retries, ) try: return _try_parse_json(raw2) except LLMJSONError as second_err: raise LLMJSONError( JSON 解析失败自修正后仍失败, raw_responseraw, corrected_responseraw2, ) from second_err