Qwen3-0.6B-FP8模型调用实战Python requests库详解与封装部署好一个AI模型只是第一步真正让它为你所用关键在于如何高效、稳定地与之对话。对于Qwen3-0.6B-FP8这类轻量级模型通过API调用是最高效的集成方式。今天我们不聊复杂的框架就聚焦于Python里最基础、最强大的requests库手把手带你从零开始打造一个能应对各种复杂场景的模型调用客户端。这篇文章不是简单的“发个请求收个回复”。我们会深入探讨几个工程实践中绕不开的难题怎么处理超长的输入文本如何优雅地中断模型的“长篇大论”在多轮对话中如何记住上下文网络抽风了怎么办最后我们会把所有零散的功能封装成一个拿来即用的Python SDK让你在项目中可以像调用本地函数一样轻松使用模型。1. 环境准备与最简调用在开始构建复杂功能之前让我们先确保基础环境就绪并完成一次最简单的成功调用。1.1 安装必备库打开你的终端或命令行执行以下命令。这里我们主要需要requests库为了后续处理JSON数据更方便也推荐安装httpx一个支持异步的现代化HTTP客户端作为备选或进阶选择。pip install requests httpx安装过程通常很快。完成后你可以在Python环境中导入它们确认没有报错。1.2 理解API端点与基础请求假设你的Qwen3-0.6B-FP8模型已经通过类似Ollama、vLLM或厂商提供的服务部署好并暴露了一个HTTP API接口。一个典型的文本生成接口可能长这样接口地址 (URL):http://localhost:8000/v1/chat/completions请求方法:POST请求头 (Headers): 需要指定内容类型为JSON。请求体 (Body): 一个JSON对象至少包含model模型名称和messages对话历史字段。下面是一个最基础的调用示例它向模型问好并打印回复。import requests import json # 1. 定义API的基础信息 API_URL http://localhost:8000/v1/chat/completions HEADERS { Content-Type: application/json } # 2. 构造请求数据 payload { model: Qwen3-0.6B-FP8, # 替换为你的实际模型名称 messages: [ {role: user, content: 你好请介绍一下你自己。} ], max_tokens: 512, # 限制模型生成的最大长度 temperature: 0.7, # 控制生成随机性0.0最确定1.0最随机 } # 3. 发送POST请求 try: response requests.post(API_URL, headersHEADERS, datajson.dumps(payload)) # 检查请求是否成功 (状态码为2xx) response.raise_for_status() # 4. 解析响应 result response.json() # 通常回复内容在 choices[0].message.content 路径下 reply result[choices][0][message][content] print(模型回复, reply) except requests.exceptions.RequestException as e: print(f请求出错{e}) except KeyError as e: print(f解析响应数据出错未找到预期字段{e}) print(原始响应, response.text)运行这段代码如果一切配置正确你应该能看到模型返回的自我介绍。这标志着你的调用链路已经打通了。2. 应对复杂场景分块、停止与上下文基础调用只能应对最简单的问答。实际应用中我们会遇到更复杂的需求。2.1 处理长文本分块与流式传输当你想让模型总结一篇很长的文档时直接发送可能超出模型上下文长度限制或导致请求超时。一个常见的策略是“分而治之”。思路将长文本按段落、句子或固定长度切分成块分别发送给模型获取分块摘要或关键信息最后再整合。这里展示一个按句子分块的简单示例。def summarize_long_text(long_text, api_url, chunk_bysentence, max_chunk_tokens500): 对长文本进行分块总结。 :param long_text: 需要总结的长文本 :param api_url: 模型API地址 :param chunk_by: 分块方式sentence或length :param max_chunk_tokens: 按长度分块时的最大token数估算 :return: 整合后的总结 summaries [] # 简单的按句号、问号、感叹号分句实际应用可能需要更健壮的分句库 if chunk_by sentence: # 这是一个非常简单的分句逻辑仅作演示 sentences [s.strip() for s in long_text.replace(。, 。\n).split(\n) if s.strip()] chunks sentences else: # 按长度分块 # 这里用一个简单的字符数估算生产环境应使用tokenizer精确计算 chunks [long_text[i:imax_chunk_tokens*2] for i in range(0, len(long_text), max_chunk_tokens*2)] print(f将文本分成了 {len(chunks)} 块进行处理。) for i, chunk in enumerate(chunks): prompt f请用一句话总结以下内容\n{chunk} payload { model: Qwen3-0.6B-FP8, messages: [{role: user, content: prompt}], max_tokens: 100, } try: resp requests.post(api_url, jsonpayload, timeout30) resp.raise_for_status() chunk_summary resp.json()[choices][0][message][content].strip() summaries.append(chunk_summary) print(f 块 {i1}/{len(chunks)} 处理完成。) except Exception as e: print(f 处理块 {i1} 时出错{e}) summaries.append(f[块{i1}总结失败]) # 将所有分块总结再次汇总 final_prompt 以下是关于同一主题的多条分句总结请将它们整合成一段连贯、全面的总结\n \n.join(summaries) final_payload { model: Qwen3-0.6B-FP8, messages: [{role: user, content: final_prompt}], max_tokens: 300, } try: final_resp requests.post(api_url, jsonfinal_payload, timeout30) final_resp.raise_for_status() return final_resp.json()[choices][0][message][content] except Exception as e: return f最终整合失败{e}。分块总结如下\n \n.join(summaries) # 使用示例 long_document 这里是你的很长很长的文档内容... # 替换为实际长文本 # final_summary summarize_long_text(long_document, API_URL) # print(最终总结, final_summary)流式传输 (Streaming)对于生成内容很长的场景另一种高级特性是流式传输。服务器会一边生成一边返回数据块客户端可以实时显示提升用户体验。这需要API支持streamTrue参数并使用response.iter_lines()来迭代读取。由于不是所有部署都支持这里先不展开但知道有这个选项很重要。2.2 控制生成停止词与停止序列有时候你希望模型在生成特定词语或达到某个条件时就停止而不是一直生成到max_tokens上限。这可以通过stop参数实现。payload_with_stop { model: Qwen3-0.6B-FP8, messages: [ {role: user, content: 请列举三种水果苹果、香蕉、} ], max_tokens: 50, temperature: 0.1, stop: [、, 。, \n] # 当模型生成“、” “。”或换行符时停止 } response requests.post(API_URL, jsonpayload_with_stop) reply response.json()[choices][0][message][content] print(reply) # 可能只生成“橙子”因为遇到“、”就停止了stop参数非常有用比如在生成列表时在最后一个项目后停止或者在生成特定格式如JSON时在闭合括号处停止确保格式正确。2.3 管理多轮对话上下文AI对话的魅力在于连续性。你需要让模型记住之前说过的话。核心在messages列表中维护完整的历史记录。每次新的请求都将之前所有的用户输入和模型回复都带上。class ConversationManager: def __init__(self, system_prompt你是一个有帮助的AI助手。): self.messages [{role: system, content: system_prompt}] def add_user_message(self, content): 添加用户消息 self.messages.append({role: user, content: content}) def add_assistant_message(self, content): 添加助手模型消息 self.messages.append({role: assistant, content: content}) def get_messages(self): 获取当前完整的对话上下文 return self.messages.copy() def chat_round(self, user_input, api_url, model_nameQwen3-0.6B-FP8): 进行一轮对话 self.add_user_message(user_input) payload { model: model_name, messages: self.get_messages(), max_tokens: 512, } try: response requests.post(api_url, jsonpayload, timeout60) response.raise_for_status() reply response.json()[choices][0][message][content] self.add_assistant_message(reply) return reply except requests.exceptions.RequestException as e: # 如果请求失败需要移除刚才添加的用户消息避免上下文不一致 self.messages.pop() raise e # 使用示例 manager ConversationManager() print(助手, manager.chat_round(今天的天气怎么样, API_URL)) print(助手, manager.chat_round(那我应该穿什么衣服出门, API_URL)) # 模型知道我们在聊天气和穿衣上下文长度限制模型有最大上下文长度如4096个tokens。当对话轮数增多messages会越来越长。一个常见的优化策略是只保留最近的N轮对话或者总结之前的对话历史后再放入上下文但这涉及更复杂的“上下文窗口管理”策略。3. 提升健壮性异常处理与重试机制网络和服务总有不稳定的时候。一个健壮的客户端必须能妥善处理错误。3.1 基础异常处理我们已经在前面的代码中使用了try...except。需要重点关注的异常包括requests.exceptions.ConnectionError: 网络连接错误。requests.exceptions.Timeout: 请求超时。requests.exceptions.HTTPError: HTTP状态码错误如404 500。json.JSONDecodeError: 响应不是有效的JSON。3.2 实现带退避的智能重试对于暂时性的网络故障或服务器过载常返回5xx错误重试是有效的策略。但简单的立即重试可能会加重服务器负担。更好的方法是“指数退避重试”。import time from requests.exceptions import RequestException, Timeout, ConnectionError def send_request_with_retry(api_url, payload, max_retries3, initial_delay1): 发送带指数退避重试机制的请求。 :param initial_delay: 初始延迟秒数 delay initial_delay for attempt in range(max_retries 1): # 1 包含第一次尝试 try: response requests.post(api_url, jsonpayload, timeout60) response.raise_for_status() # 如果状态码不是2xx会抛出HTTPError return response # 成功则直接返回 except (Timeout, ConnectionError) as e: # 这些是典型的可重试错误 if attempt max_retries: print(f请求失败已达最大重试次数 {max_retries}。) raise e print(f请求遇到临时错误 ({e}) {delay}秒后重试... (尝试 {attempt 1}/{max_retries})) time.sleep(delay) delay * 2 # 指数退避 except RequestException as e: # 其他请求异常如HTTP 4xx错误通常不可重试 print(f请求发生不可重试错误: {e}) raise e # 理论上不会执行到这里 raise Exception(重试逻辑异常) # 使用示例 try: payload {model: Qwen3-0.6B-FP8, messages: [{role: user, content: Hello}]} resp send_request_with_retry(API_URL, payload, max_retries3) print(请求成功, resp.json()) except Exception as e: print(最终请求失败, e)这个函数会在遇到网络超时或连接错误时自动重试每次重试前等待的时间翻倍给服务器恢复的时间。4. 终极封装打造你的模型客户端SDK现在我们把上面所有的功能——基础调用、上下文管理、异常重试——整合到一个类里创建一个功能完整、易于使用的客户端。import requests import json import time from typing import List, Dict, Optional, Any class QwenClient: Qwen模型API的Python客户端封装。 def __init__(self, base_url: str, model_name: str Qwen3-0.6B-FP8, timeout: int 60, max_retries: int 3): 初始化客户端。 :param base_url: API基础地址例如 http://localhost:8000/v1 :param model_name: 模型名称 :param timeout: 请求超时时间秒 :param max_retries: 最大重试次数 self.base_url base_url.rstrip(/) self.chat_completion_url f{self.base_url}/chat/completions self.model_name model_name self.timeout timeout self.max_retries max_retries self.conversation_history [] def _send_request(self, payload: Dict) - Dict: 内部方法发送请求并处理重试逻辑。 delay 1 headers {Content-Type: application/json} for attempt in range(self.max_retries 1): try: response requests.post( self.chat_completion_url, headersheaders, datajson.dumps(payload), timeoutself.timeout ) response.raise_for_status() return response.json() except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: if attempt self.max_retries: raise Exception(f请求失败已重试{self.max_retries}次。最后错误: {e}) print(f请求暂时失败 ({e}) {delay}秒后重试...) time.sleep(delay) delay * 2 except requests.exceptions.RequestException as e: # 对于HTTP 4xx等错误通常不重试 raise Exception(f请求发生错误: {e}) def chat(self, user_message: str, system_prompt: Optional[str] None, temperature: float 0.7, max_tokens: int 512, stop: Optional[List[str]] None, stream: bool False) - str: 发送单轮或多轮对话消息。 :param user_message: 用户输入的消息 :param system_prompt: 系统提示词如果为None则使用历史记录中的或默认 :return: 模型的回复文本 # 构建本次请求的messages messages_for_this_request [] # 1. 添加系统提示如果本次指定了或历史记录中没有系统消息 if system_prompt: # 如果指定了新的系统提示我们将其放在本次请求的开头。 # 注意更复杂的实现可能需要管理历史中的系统消息。 messages_for_this_request.append({role: system, content: system_prompt}) # 简单起见这里清空历史以新的系统提示开始。实际可根据需求调整。 self.conversation_history [{role: system, content: system_prompt}] elif not self.conversation_history or self.conversation_history[0][role] ! system: # 如果历史为空或第一个不是系统消息添加一个默认的 default_sys 你是一个有帮助的AI助手。 messages_for_this_request.append({role: system, content: default_sys}) self.conversation_history [{role: system, content: default_sys}] # 2. 添加历史对话除了可能被覆盖的系统消息 # 这里简单地将当前历史记录可能已包含系统消息全部加入 messages_for_this_request self.conversation_history.copy() # 3. 添加本次用户消息 messages_for_this_request.append({role: user, content: user_message}) # 4. 构建请求载荷 payload { model: self.model_name, messages: messages_for_this_request, temperature: temperature, max_tokens: max_tokens, } if stop: payload[stop] stop if stream: payload[stream] True # 流式处理需要更复杂的逻辑此处暂不实现 raise NotImplementedError(流式处理功能暂未在本示例中实现。) # 5. 发送请求 try: response_data self._send_request(payload) assistant_reply response_data[choices][0][message][content] # 6. 更新历史记录 # 确保历史记录里已经有最新的用户消息和模型回复 if messages_for_this_request[-1][role] user and messages_for_this_request[-1][content] user_message: # 如果本次请求的messages末尾已经是当前用户消息则只需添加助手回复 self.conversation_history.append({role: user, content: user_message}) self.conversation_history.append({role: assistant, content: assistant_reply}) else: # 否则用本次完整的messages更新历史这种情况发生在system_prompt被重置时 self.conversation_history messages_for_this_request self.conversation_history.append({role: assistant, content: assistant_reply}) return assistant_reply except Exception as e: # 请求失败不更新历史记录中的用户消息 print(f对话请求失败: {e}) raise e def clear_history(self): 清空对话历史。 self.conversation_history [] def get_history(self) - List[Dict]: 获取当前对话历史。 return self.conversation_history.copy() # 使用示例 if __name__ __main__: # 1. 初始化客户端 client QwenClient(base_urlhttp://localhost:8000/v1, model_nameQwen3-0.6B-FP8) # 2. 进行多轮对话 print(用户你好) reply1 client.chat(你好) print(f助手{reply1}\n) print(用户你能做什么) reply2 client.chat(你能做什么) print(f助手{reply2}\n) # 3. 查看历史 print(当前对话历史) for msg in client.get_history(): print(f {msg[role]}: {msg[content][:50]}...) # 4. 清空历史开始新话题 client.clear_history() reply3 client.chat(我们来聊聊科学。, system_prompt你是一个严谨的科学家。) print(f\n新对话助手{reply3})这个QwenClient类提供了一个简洁的接口。你只需要初始化一次然后通过chat方法进行对话它会自动帮你管理上下文、处理重试和异常。你可以根据需要继续为它添加流式支持、函数调用、更精细的上下文窗口管理等功能。5. 总结与建议走完这一趟你应该不再满足于仅仅发送一个简单的POST请求了。我们探讨了如何用requests库处理长文本、控制生成过程、维护对话状态并最终构建了一个具备基本生产可用性的客户端。实际使用中还有几点值得注意一是Token计算对于长文本最好使用模型对应的tokenizer来精确计算长度避免超出限制。二是异步处理如果你的应用是高并发的可以考虑使用httpx或aiohttp库进行异步调用提升效率。三是配置管理将API地址、密钥等敏感信息放在环境变量或配置文件中而不是硬编码在代码里。封装好的QwenClient只是一个起点。你可以根据项目的具体需求为它添加日志记录、性能监控、缓存层等更多功能。希望这份详实的指南能让你在调用Qwen或其他大模型API时更加得心应手。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
Qwen3-0.6B-FP8模型调用实战：Python requests库详解与封装
Qwen3-0.6B-FP8模型调用实战Python requests库详解与封装部署好一个AI模型只是第一步真正让它为你所用关键在于如何高效、稳定地与之对话。对于Qwen3-0.6B-FP8这类轻量级模型通过API调用是最高效的集成方式。今天我们不聊复杂的框架就聚焦于Python里最基础、最强大的requests库手把手带你从零开始打造一个能应对各种复杂场景的模型调用客户端。这篇文章不是简单的“发个请求收个回复”。我们会深入探讨几个工程实践中绕不开的难题怎么处理超长的输入文本如何优雅地中断模型的“长篇大论”在多轮对话中如何记住上下文网络抽风了怎么办最后我们会把所有零散的功能封装成一个拿来即用的Python SDK让你在项目中可以像调用本地函数一样轻松使用模型。1. 环境准备与最简调用在开始构建复杂功能之前让我们先确保基础环境就绪并完成一次最简单的成功调用。1.1 安装必备库打开你的终端或命令行执行以下命令。这里我们主要需要requests库为了后续处理JSON数据更方便也推荐安装httpx一个支持异步的现代化HTTP客户端作为备选或进阶选择。pip install requests httpx安装过程通常很快。完成后你可以在Python环境中导入它们确认没有报错。1.2 理解API端点与基础请求假设你的Qwen3-0.6B-FP8模型已经通过类似Ollama、vLLM或厂商提供的服务部署好并暴露了一个HTTP API接口。一个典型的文本生成接口可能长这样接口地址 (URL):http://localhost:8000/v1/chat/completions请求方法:POST请求头 (Headers): 需要指定内容类型为JSON。请求体 (Body): 一个JSON对象至少包含model模型名称和messages对话历史字段。下面是一个最基础的调用示例它向模型问好并打印回复。import requests import json # 1. 定义API的基础信息 API_URL http://localhost:8000/v1/chat/completions HEADERS { Content-Type: application/json } # 2. 构造请求数据 payload { model: Qwen3-0.6B-FP8, # 替换为你的实际模型名称 messages: [ {role: user, content: 你好请介绍一下你自己。} ], max_tokens: 512, # 限制模型生成的最大长度 temperature: 0.7, # 控制生成随机性0.0最确定1.0最随机 } # 3. 发送POST请求 try: response requests.post(API_URL, headersHEADERS, datajson.dumps(payload)) # 检查请求是否成功 (状态码为2xx) response.raise_for_status() # 4. 解析响应 result response.json() # 通常回复内容在 choices[0].message.content 路径下 reply result[choices][0][message][content] print(模型回复, reply) except requests.exceptions.RequestException as e: print(f请求出错{e}) except KeyError as e: print(f解析响应数据出错未找到预期字段{e}) print(原始响应, response.text)运行这段代码如果一切配置正确你应该能看到模型返回的自我介绍。这标志着你的调用链路已经打通了。2. 应对复杂场景分块、停止与上下文基础调用只能应对最简单的问答。实际应用中我们会遇到更复杂的需求。2.1 处理长文本分块与流式传输当你想让模型总结一篇很长的文档时直接发送可能超出模型上下文长度限制或导致请求超时。一个常见的策略是“分而治之”。思路将长文本按段落、句子或固定长度切分成块分别发送给模型获取分块摘要或关键信息最后再整合。这里展示一个按句子分块的简单示例。def summarize_long_text(long_text, api_url, chunk_bysentence, max_chunk_tokens500): 对长文本进行分块总结。 :param long_text: 需要总结的长文本 :param api_url: 模型API地址 :param chunk_by: 分块方式sentence或length :param max_chunk_tokens: 按长度分块时的最大token数估算 :return: 整合后的总结 summaries [] # 简单的按句号、问号、感叹号分句实际应用可能需要更健壮的分句库 if chunk_by sentence: # 这是一个非常简单的分句逻辑仅作演示 sentences [s.strip() for s in long_text.replace(。, 。\n).split(\n) if s.strip()] chunks sentences else: # 按长度分块 # 这里用一个简单的字符数估算生产环境应使用tokenizer精确计算 chunks [long_text[i:imax_chunk_tokens*2] for i in range(0, len(long_text), max_chunk_tokens*2)] print(f将文本分成了 {len(chunks)} 块进行处理。) for i, chunk in enumerate(chunks): prompt f请用一句话总结以下内容\n{chunk} payload { model: Qwen3-0.6B-FP8, messages: [{role: user, content: prompt}], max_tokens: 100, } try: resp requests.post(api_url, jsonpayload, timeout30) resp.raise_for_status() chunk_summary resp.json()[choices][0][message][content].strip() summaries.append(chunk_summary) print(f 块 {i1}/{len(chunks)} 处理完成。) except Exception as e: print(f 处理块 {i1} 时出错{e}) summaries.append(f[块{i1}总结失败]) # 将所有分块总结再次汇总 final_prompt 以下是关于同一主题的多条分句总结请将它们整合成一段连贯、全面的总结\n \n.join(summaries) final_payload { model: Qwen3-0.6B-FP8, messages: [{role: user, content: final_prompt}], max_tokens: 300, } try: final_resp requests.post(api_url, jsonfinal_payload, timeout30) final_resp.raise_for_status() return final_resp.json()[choices][0][message][content] except Exception as e: return f最终整合失败{e}。分块总结如下\n \n.join(summaries) # 使用示例 long_document 这里是你的很长很长的文档内容... # 替换为实际长文本 # final_summary summarize_long_text(long_document, API_URL) # print(最终总结, final_summary)流式传输 (Streaming)对于生成内容很长的场景另一种高级特性是流式传输。服务器会一边生成一边返回数据块客户端可以实时显示提升用户体验。这需要API支持streamTrue参数并使用response.iter_lines()来迭代读取。由于不是所有部署都支持这里先不展开但知道有这个选项很重要。2.2 控制生成停止词与停止序列有时候你希望模型在生成特定词语或达到某个条件时就停止而不是一直生成到max_tokens上限。这可以通过stop参数实现。payload_with_stop { model: Qwen3-0.6B-FP8, messages: [ {role: user, content: 请列举三种水果苹果、香蕉、} ], max_tokens: 50, temperature: 0.1, stop: [、, 。, \n] # 当模型生成“、” “。”或换行符时停止 } response requests.post(API_URL, jsonpayload_with_stop) reply response.json()[choices][0][message][content] print(reply) # 可能只生成“橙子”因为遇到“、”就停止了stop参数非常有用比如在生成列表时在最后一个项目后停止或者在生成特定格式如JSON时在闭合括号处停止确保格式正确。2.3 管理多轮对话上下文AI对话的魅力在于连续性。你需要让模型记住之前说过的话。核心在messages列表中维护完整的历史记录。每次新的请求都将之前所有的用户输入和模型回复都带上。class ConversationManager: def __init__(self, system_prompt你是一个有帮助的AI助手。): self.messages [{role: system, content: system_prompt}] def add_user_message(self, content): 添加用户消息 self.messages.append({role: user, content: content}) def add_assistant_message(self, content): 添加助手模型消息 self.messages.append({role: assistant, content: content}) def get_messages(self): 获取当前完整的对话上下文 return self.messages.copy() def chat_round(self, user_input, api_url, model_nameQwen3-0.6B-FP8): 进行一轮对话 self.add_user_message(user_input) payload { model: model_name, messages: self.get_messages(), max_tokens: 512, } try: response requests.post(api_url, jsonpayload, timeout60) response.raise_for_status() reply response.json()[choices][0][message][content] self.add_assistant_message(reply) return reply except requests.exceptions.RequestException as e: # 如果请求失败需要移除刚才添加的用户消息避免上下文不一致 self.messages.pop() raise e # 使用示例 manager ConversationManager() print(助手, manager.chat_round(今天的天气怎么样, API_URL)) print(助手, manager.chat_round(那我应该穿什么衣服出门, API_URL)) # 模型知道我们在聊天气和穿衣上下文长度限制模型有最大上下文长度如4096个tokens。当对话轮数增多messages会越来越长。一个常见的优化策略是只保留最近的N轮对话或者总结之前的对话历史后再放入上下文但这涉及更复杂的“上下文窗口管理”策略。3. 提升健壮性异常处理与重试机制网络和服务总有不稳定的时候。一个健壮的客户端必须能妥善处理错误。3.1 基础异常处理我们已经在前面的代码中使用了try...except。需要重点关注的异常包括requests.exceptions.ConnectionError: 网络连接错误。requests.exceptions.Timeout: 请求超时。requests.exceptions.HTTPError: HTTP状态码错误如404 500。json.JSONDecodeError: 响应不是有效的JSON。3.2 实现带退避的智能重试对于暂时性的网络故障或服务器过载常返回5xx错误重试是有效的策略。但简单的立即重试可能会加重服务器负担。更好的方法是“指数退避重试”。import time from requests.exceptions import RequestException, Timeout, ConnectionError def send_request_with_retry(api_url, payload, max_retries3, initial_delay1): 发送带指数退避重试机制的请求。 :param initial_delay: 初始延迟秒数 delay initial_delay for attempt in range(max_retries 1): # 1 包含第一次尝试 try: response requests.post(api_url, jsonpayload, timeout60) response.raise_for_status() # 如果状态码不是2xx会抛出HTTPError return response # 成功则直接返回 except (Timeout, ConnectionError) as e: # 这些是典型的可重试错误 if attempt max_retries: print(f请求失败已达最大重试次数 {max_retries}。) raise e print(f请求遇到临时错误 ({e}) {delay}秒后重试... (尝试 {attempt 1}/{max_retries})) time.sleep(delay) delay * 2 # 指数退避 except RequestException as e: # 其他请求异常如HTTP 4xx错误通常不可重试 print(f请求发生不可重试错误: {e}) raise e # 理论上不会执行到这里 raise Exception(重试逻辑异常) # 使用示例 try: payload {model: Qwen3-0.6B-FP8, messages: [{role: user, content: Hello}]} resp send_request_with_retry(API_URL, payload, max_retries3) print(请求成功, resp.json()) except Exception as e: print(最终请求失败, e)这个函数会在遇到网络超时或连接错误时自动重试每次重试前等待的时间翻倍给服务器恢复的时间。4. 终极封装打造你的模型客户端SDK现在我们把上面所有的功能——基础调用、上下文管理、异常重试——整合到一个类里创建一个功能完整、易于使用的客户端。import requests import json import time from typing import List, Dict, Optional, Any class QwenClient: Qwen模型API的Python客户端封装。 def __init__(self, base_url: str, model_name: str Qwen3-0.6B-FP8, timeout: int 60, max_retries: int 3): 初始化客户端。 :param base_url: API基础地址例如 http://localhost:8000/v1 :param model_name: 模型名称 :param timeout: 请求超时时间秒 :param max_retries: 最大重试次数 self.base_url base_url.rstrip(/) self.chat_completion_url f{self.base_url}/chat/completions self.model_name model_name self.timeout timeout self.max_retries max_retries self.conversation_history [] def _send_request(self, payload: Dict) - Dict: 内部方法发送请求并处理重试逻辑。 delay 1 headers {Content-Type: application/json} for attempt in range(self.max_retries 1): try: response requests.post( self.chat_completion_url, headersheaders, datajson.dumps(payload), timeoutself.timeout ) response.raise_for_status() return response.json() except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: if attempt self.max_retries: raise Exception(f请求失败已重试{self.max_retries}次。最后错误: {e}) print(f请求暂时失败 ({e}) {delay}秒后重试...) time.sleep(delay) delay * 2 except requests.exceptions.RequestException as e: # 对于HTTP 4xx等错误通常不重试 raise Exception(f请求发生错误: {e}) def chat(self, user_message: str, system_prompt: Optional[str] None, temperature: float 0.7, max_tokens: int 512, stop: Optional[List[str]] None, stream: bool False) - str: 发送单轮或多轮对话消息。 :param user_message: 用户输入的消息 :param system_prompt: 系统提示词如果为None则使用历史记录中的或默认 :return: 模型的回复文本 # 构建本次请求的messages messages_for_this_request [] # 1. 添加系统提示如果本次指定了或历史记录中没有系统消息 if system_prompt: # 如果指定了新的系统提示我们将其放在本次请求的开头。 # 注意更复杂的实现可能需要管理历史中的系统消息。 messages_for_this_request.append({role: system, content: system_prompt}) # 简单起见这里清空历史以新的系统提示开始。实际可根据需求调整。 self.conversation_history [{role: system, content: system_prompt}] elif not self.conversation_history or self.conversation_history[0][role] ! system: # 如果历史为空或第一个不是系统消息添加一个默认的 default_sys 你是一个有帮助的AI助手。 messages_for_this_request.append({role: system, content: default_sys}) self.conversation_history [{role: system, content: default_sys}] # 2. 添加历史对话除了可能被覆盖的系统消息 # 这里简单地将当前历史记录可能已包含系统消息全部加入 messages_for_this_request self.conversation_history.copy() # 3. 添加本次用户消息 messages_for_this_request.append({role: user, content: user_message}) # 4. 构建请求载荷 payload { model: self.model_name, messages: messages_for_this_request, temperature: temperature, max_tokens: max_tokens, } if stop: payload[stop] stop if stream: payload[stream] True # 流式处理需要更复杂的逻辑此处暂不实现 raise NotImplementedError(流式处理功能暂未在本示例中实现。) # 5. 发送请求 try: response_data self._send_request(payload) assistant_reply response_data[choices][0][message][content] # 6. 更新历史记录 # 确保历史记录里已经有最新的用户消息和模型回复 if messages_for_this_request[-1][role] user and messages_for_this_request[-1][content] user_message: # 如果本次请求的messages末尾已经是当前用户消息则只需添加助手回复 self.conversation_history.append({role: user, content: user_message}) self.conversation_history.append({role: assistant, content: assistant_reply}) else: # 否则用本次完整的messages更新历史这种情况发生在system_prompt被重置时 self.conversation_history messages_for_this_request self.conversation_history.append({role: assistant, content: assistant_reply}) return assistant_reply except Exception as e: # 请求失败不更新历史记录中的用户消息 print(f对话请求失败: {e}) raise e def clear_history(self): 清空对话历史。 self.conversation_history [] def get_history(self) - List[Dict]: 获取当前对话历史。 return self.conversation_history.copy() # 使用示例 if __name__ __main__: # 1. 初始化客户端 client QwenClient(base_urlhttp://localhost:8000/v1, model_nameQwen3-0.6B-FP8) # 2. 进行多轮对话 print(用户你好) reply1 client.chat(你好) print(f助手{reply1}\n) print(用户你能做什么) reply2 client.chat(你能做什么) print(f助手{reply2}\n) # 3. 查看历史 print(当前对话历史) for msg in client.get_history(): print(f {msg[role]}: {msg[content][:50]}...) # 4. 清空历史开始新话题 client.clear_history() reply3 client.chat(我们来聊聊科学。, system_prompt你是一个严谨的科学家。) print(f\n新对话助手{reply3})这个QwenClient类提供了一个简洁的接口。你只需要初始化一次然后通过chat方法进行对话它会自动帮你管理上下文、处理重试和异常。你可以根据需要继续为它添加流式支持、函数调用、更精细的上下文窗口管理等功能。5. 总结与建议走完这一趟你应该不再满足于仅仅发送一个简单的POST请求了。我们探讨了如何用requests库处理长文本、控制生成过程、维护对话状态并最终构建了一个具备基本生产可用性的客户端。实际使用中还有几点值得注意一是Token计算对于长文本最好使用模型对应的tokenizer来精确计算长度避免超出限制。二是异步处理如果你的应用是高并发的可以考虑使用httpx或aiohttp库进行异步调用提升效率。三是配置管理将API地址、密钥等敏感信息放在环境变量或配置文件中而不是硬编码在代码里。封装好的QwenClient只是一个起点。你可以根据项目的具体需求为它添加日志记录、性能监控、缓存层等更多功能。希望这份详实的指南能让你在调用Qwen或其他大模型API时更加得心应手。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
