InternLM2-Chat-1.8B快速调用API详解Python客户端与流式响应处理1. 引言如果你已经成功在星图GPU平台上部署了InternLM2-Chat-1.8B模型接下来最关心的问题可能就是我该怎么用程序调用它是每次都要等它全部生成完才能看到结果还是可以像聊天一样看着文字一个字一个字地蹦出来这篇文章就是为你准备的。我们不谈复杂的模型原理也不讲高深的架构设计就聚焦一件事用Python写一个简单好用的客户端去调用那个已经部署好的API。重点是我们要实现“流式响应”——也就是让模型生成文字的过程实时显示出来就像你在和真人对话一样能看到对方正在输入。为什么这很重要想象一下你让模型写一篇几百字的文章如果等它全部生成完再一次性返回你可能要等上十几秒甚至更久期间完全不知道进展如何甚至怀疑程序是不是卡住了。而流式响应能让你实时看到生成的内容不仅体验更好还能在发现方向不对时及时中断。我会带你从最基础的HTTP请求开始一步步构建一个功能完整的客户端包括错误处理和重试机制。即使你之前没怎么用过requests库跟着做下来也能轻松掌握。2. 环境准备与快速开始在开始写代码之前我们需要确保环境里有所需的工具。整个过程很简单基本上就是安装一个Python库。2.1 安装必要的库核心库只有一个requests。它是Python里最常用的HTTP库用起来非常直观。打开你的终端或命令行输入下面这行命令pip install requests如果你用的是Python 3可能需要用pip3代替pip。安装过程通常很快完成后可以验证一下import requests print(requests.__version__)能正常输出版本号比如2.31.0就说明安装成功了。2.2 获取API访问信息要调用API你需要知道三样东西API地址Endpoint这是你部署模型后星图平台提供给你的URL通常以http://或https://开头。认证方式可能是API Key也可能是Token具体看你的部署配置。星图平台一般会在控制台明确给出。模型名称这里就是InternLM2-Chat-1.8B但有时API接口可能只需要你指定路径模型名已经包含在URL里。把这些信息准备好记在一个方便的地方我们接下来写代码时会用到。假设你的API地址是https://your-deployment-url/v1/chat/completions有一个API Key是sk-xxxxxx。3. 基础API调用从第一行代码开始我们先从最简单的开始发送一个请求然后等待完整的回复。这种方式虽然不“流式”但能让我们先理解整个交互流程。3.1 构建你的第一个请求API调用本质上就是向一个特定的网址发送一段格式化的数据JSON然后接收返回的数据。对于聊天模型最常见的格式是模仿OpenAI的Chat Completions接口。下面是一个最基础的示例代码import requests import json # 你的API信息 API_URL https://your-deployment-url/v1/chat/completions API_KEY sk-xxxxxx # 请替换为你的真实Key # 准备请求头通常包含认证信息和内容类型 headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } # 准备请求体告诉模型你想聊什么 payload { model: InternLM2-Chat-1.8B, # 模型名称根据实际情况调整 messages: [ {role: user, content: 你好请介绍一下你自己。} ], max_tokens: 512, # 限制生成的最大长度 temperature: 0.7, # 控制随机性0.0最确定1.0最随机 } # 发送POST请求 response requests.post(API_URL, headersheaders, jsonpayload) # 检查请求是否成功 if response.status_code 200: # 解析返回的JSON数据 result response.json() # 提取模型回复的内容 reply result[choices][0][message][content] print(模型回复, reply) else: print(f请求失败状态码{response.status_code}) print(f错误信息{response.text})把代码里的API_URL和API_KEY换成你自己的然后运行。如果一切正常你应该能看到InternLM2-Chat-1.8B模型的自我介绍。代码要点解析headers里的Authorization是认证的关键Bearer后面跟着你的API Key。messages是一个列表里面按顺序存放对话历史。每条消息都有role角色如user或assistant和content内容。通过这个列表模型就能理解对话的上下文。max_tokens和temperature是两个常用的参数用来控制生成文本的长度和创造性。3.2 处理多轮对话聊天机器人好玩的地方在于能记住上下文。我们只需要在messages列表里依次追加对话历史就行。# 续接上面的代码假设我们已经有了第一次的回复 first_reply # 构建第二轮对话的请求 payload { model: InternLM2-Chat-1.8B, messages: [ {role: user, content: 你好请介绍一下你自己。}, {role: assistant, content: first_reply}, # 上一轮模型的回复 {role: user, content: 你刚才说的很好能再详细说说你的训练数据吗} # 用户的新问题 ], max_tokens: 512, temperature: 0.7, } # 再次发送请求 response requests.post(API_URL, headersheaders, jsonpayload) # ... 处理回复你看只要把之前的对话记录都按顺序放进messages里模型就能自然地接上话茬实现连续对话。4. 实现流式响应让文字“流”起来基础调用虽然简单但体验上差点意思。现在我们来实现更酷的流式响应。4.1 什么是流式响应普通请求就像发一封邮件你写好了寄出去然后等对方回一封完整的信。流式响应则像打电话对方一边说你一边就能听到。在技术层面开启流式响应后服务器不会等所有内容都生成完再一次性发送给你而是每生成一小段比如一个词或一句话就立刻把这一小段数据发过来。客户端收到一段就显示一段从而实现实时的打字机效果。4.2 开启流式请求实现流式响应的关键是在请求参数里加一个stream: true。同时我们不能再使用普通的requests.post()然后等待全部返回而需要处理一个持续的数据流。import requests import json API_URL https://your-deployment-url/v1/chat/completions API_KEY sk-xxxxxx headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } payload { model: InternLM2-Chat-1.8B, messages: [{role: user, content: 写一个关于人工智能的短故事大约100字。}], max_tokens: 200, temperature: 0.8, stream: True # 关键告诉服务器我们要流式输出 } # 注意这里streamTrue 参数告诉requests库我们要处理流 response requests.post(API_URL, headersheaders, jsonpayload, streamTrue) # 检查初始响应是否成功 if response.status_code ! 200: print(f请求失败: {response.status_code}) print(response.text) else: print(开始接收流式响应) collected_content # 迭代处理响应的每一行 for line in response.iter_lines(): if line: # 流式响应通常以 data: 开头 decoded_line line.decode(utf-8) if decoded_line.startswith(data: ): data_str decoded_line[6:] # 去掉 data: 前缀 if data_str [DONE]: # 流结束的标志 print(\n\n--- 生成完成 ---) break try: data json.loads(data_str) # 提取当前生成的一小段文本 delta_content data[choices][0][delta].get(content, ) if delta_content: print(delta_content, end, flushTrue) # 关键即时打印不换行 collected_content delta_content except json.JSONDecodeError: # 有时可能会收到非JSON数据如心跳包忽略即可 pass print(f\n\n完整内容\n{collected_content})运行这段代码你会看到故事是一个词一个词或一句话一句话地显示出来的而不是等了好久突然出现一整段。代码核心解析stream: True这是发给服务器的指令。requests.post(..., streamTrue)这是告诉requests库以流模式处理连接。response.iter_lines()这是一个迭代器它会持续读取从服务器发来的数据流每次返回一行。服务器发送的每一行数据通常以data:开头后面跟着一个JSON对象或[DONE]。我们从JSON对象的choices[0].delta.content路径下取出最新生成的一小段文本并立即打印出来。flushTrue确保内容立刻显示在屏幕上。4.3 构建一个更健壮的流式客户端上面的例子展示了基本原理但还不够健壮。我们把它封装成一个函数并加入更好的错误处理和状态管理。import requests import json import sys class StreamingLLMClient: def __init__(self, api_url, api_key): self.api_url api_url self.headers { Authorization: fBearer {api_key}, Content-Type: application/json } def chat_stream(self, messages, max_tokens512, temperature0.7): 发起流式聊天请求并实时打印结果 payload { model: InternLM2-Chat-1.8B, messages: messages, max_tokens: max_tokens, temperature: temperature, stream: True } try: response requests.post(self.api_url, headersself.headers, jsonpayload, streamTrue, timeout60) response.raise_for_status() # 如果状态码不是200会抛出异常 except requests.exceptions.RequestException as e: print(f\n请求发生错误: {e}) return None full_content print(模型正在思考..., end\n\n) try: for line in response.iter_lines(decode_unicodeTrue): if line: if line.startswith(data: ): data_str line[6:] if data_str [DONE]: print(\n[流结束]) break try: data json.loads(data_str) delta data.get(choices, [{}])[0].get(delta, {}) chunk delta.get(content, ) if chunk: print(chunk, end, flushTrue) full_content chunk except json.JSONDecodeError: # 忽略非JSON数据如心跳包 continue except KeyboardInterrupt: print(\n\n用户中断了生成。) return full_content except Exception as e: print(f\n处理数据流时发生错误: {e}) return full_content return full_content # 使用示例 if __name__ __main__: client StreamingLLMClient( api_urlhttps://your-deployment-url/v1/chat/completions, api_keysk-xxxxxx ) history [ {role: user, content: 用Python写一个简单的函数计算斐波那契数列的前n项。} ] reply client.chat_stream(history, max_tokens300) if reply: # 你可以把完整的回复存下来用于后续的多轮对话 history.append({role: assistant, content: reply}) print(f\n\n完整回复已保存共{len(reply)}字符。)这个类更完善一些它处理了网络请求异常、JSON解析异常还优雅地处理了用户中途按CtrlC中断的情况。decode_unicodeTrue参数让iter_lines直接返回字符串省去了手动解码的步骤。5. 错误处理与重试机制网络请求总有可能出问题服务器暂时不可用、网络波动、请求超时等等。一个健壮的客户端必须能妥善处理这些情况。5.1 常见的错误类型及处理认证错误401/403API Key错了或者没有权限。处理方式检查Key提示用户。请求格式错误400你发送的JSON数据格式不对比如少了必需的字段。处理方式检查payload结构查看错误信息。服务器错误5xx服务器端出了问题。处理方式等待一会儿重试。网络超时或连接错误网络不稳定。处理方式重试。5.2 为流式请求添加重试机制对于流式请求重试稍微复杂一点因为连接可能在中途断开。一个简单的策略是如果连接在内容生成完成前断开就重新发起请求并附带上之前已经生成好的部分内容作为上下文。下面是一个带有简单重试功能的流式对话函数import time import requests import json def robust_chat_stream_with_retry(api_url, api_key, messages, max_retries3): 带有重试机制的流式聊天函数 headers { Authorization: fBearer {api_key}, Content-Type: application/json } payload_base { model: InternLM2-Chat-1.8B, max_tokens: 512, temperature: 0.7, stream: True } full_content attempt 0 while attempt max_retries: attempt 1 print(f\n尝试第 {attempt} 次请求...) # 每次重试都使用完整的对话历史包括已生成的部分 current_messages messages.copy() if full_content: # 如果之前已经生成了一部分把它作为assistant的回复加入历史 # 这样重试时模型就知道从哪里继续 current_messages.append({role: assistant, content: full_content}) # 注意这里我们模拟用户说“继续”也可以根据场景调整提示 current_messages.append({role: user, content: 请继续。}) payload {**payload_base, messages: current_messages} try: response requests.post(api_url, headersheaders, jsonpayload, streamTrue, timeout30) response.raise_for_status() # 处理流式响应 for line in response.iter_lines(decode_unicodeTrue): if line and line.startswith(data: ): data_str line[6:] if data_str [DONE]: print(\n[生成完成]) return full_content # 成功完成返回内容 try: data json.loads(data_str) chunk data.get(choices, [{}])[0].get(delta, {}).get(content, ) if chunk: print(chunk, end, flushTrue) full_content chunk except json.JSONDecodeError: continue # 如果正常循环结束没有遇到[DONE]可能是流提前结束 print(\n[流意外结束]) # 如果已经生成了一些内容可以考虑返回 if full_content: return full_content else: # 否则准备重试 time.sleep(2) # 等待2秒后重试 continue except (requests.exceptions.Timeout, requests.exceptions.ConnectionError, requests.exceptions.ChunkedEncodingError) as e: print(f\n网络错误: {e}) if attempt max_retries: wait_time 2 ** attempt # 指数退避2, 4, 8秒... print(f等待 {wait_time} 秒后重试...) time.sleep(wait_time) else: print(已达到最大重试次数。) return full_content # 返回已生成的部分 except requests.exceptions.HTTPError as e: print(f\nHTTP错误 (状态码: {e.response.status_code}): {e.response.text[:200]}) # 如果是客户端错误4xx重试可能没用 if 400 e.response.status_code 500: return full_content else: if attempt max_retries: time.sleep(2) continue else: return full_content except Exception as e: print(f\n未知错误: {e}) return full_content return full_content # 使用示例 if __name__ __main__: API_URL https://your-deployment-url/v1/chat/completions API_KEY sk-xxxxxx conversation [ {role: user, content: 给我讲一个长篇的科幻故事开头。} ] result robust_chat_stream_with_retry(API_URL, API_KEY, conversation, max_retries3) print(f\n\n最终获取的内容长度: {len(result)} 字符)这个函数实现了几个关键点指数退避重试每次重试前等待的时间逐渐增加2秒、4秒、8秒避免对服务器造成压力。上下文恢复如果流中断前已经生成了一部分内容重试时会把这部分内容作为对话历史发送回去并提示模型“继续”从而尽量保持内容的连贯性。错误分类处理对网络错误可重试和客户端错误如400 Bad Request重试可能无用区别对待。6. 总结走完这一趟你应该已经掌握了用Python调用InternLM2-Chat-1.8B模型API的核心方法。从最基础的发送一个请求并等待回复到实现实时的流式响应看着文字一个个蹦出来再到为你的客户端穿上“防弹衣”加上错误处理和重试机制每一步都是在让这个工具变得更实用、更可靠。流式响应的体验提升是巨大的尤其是生成长文本时那种即时反馈的感觉完全不同于干等着。而良好的错误处理则能让你的程序在不太稳定的网络环境下依然保持可用性不至于因为一次短暂的波动就彻底崩溃。在实际使用中你可以根据需求继续扩展这个客户端比如加入对话历史管理、实现函数调用如果模型支持、或者做一个简单的图形界面。核心的通信逻辑已经在这里了。希望这个教程能帮你快速上手把部署好的模型真正用起来做出一些有趣或者有用的应用。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
InternLM2-Chat-1.8B快速调用API详解：Python客户端与流式响应处理
InternLM2-Chat-1.8B快速调用API详解Python客户端与流式响应处理1. 引言如果你已经成功在星图GPU平台上部署了InternLM2-Chat-1.8B模型接下来最关心的问题可能就是我该怎么用程序调用它是每次都要等它全部生成完才能看到结果还是可以像聊天一样看着文字一个字一个字地蹦出来这篇文章就是为你准备的。我们不谈复杂的模型原理也不讲高深的架构设计就聚焦一件事用Python写一个简单好用的客户端去调用那个已经部署好的API。重点是我们要实现“流式响应”——也就是让模型生成文字的过程实时显示出来就像你在和真人对话一样能看到对方正在输入。为什么这很重要想象一下你让模型写一篇几百字的文章如果等它全部生成完再一次性返回你可能要等上十几秒甚至更久期间完全不知道进展如何甚至怀疑程序是不是卡住了。而流式响应能让你实时看到生成的内容不仅体验更好还能在发现方向不对时及时中断。我会带你从最基础的HTTP请求开始一步步构建一个功能完整的客户端包括错误处理和重试机制。即使你之前没怎么用过requests库跟着做下来也能轻松掌握。2. 环境准备与快速开始在开始写代码之前我们需要确保环境里有所需的工具。整个过程很简单基本上就是安装一个Python库。2.1 安装必要的库核心库只有一个requests。它是Python里最常用的HTTP库用起来非常直观。打开你的终端或命令行输入下面这行命令pip install requests如果你用的是Python 3可能需要用pip3代替pip。安装过程通常很快完成后可以验证一下import requests print(requests.__version__)能正常输出版本号比如2.31.0就说明安装成功了。2.2 获取API访问信息要调用API你需要知道三样东西API地址Endpoint这是你部署模型后星图平台提供给你的URL通常以http://或https://开头。认证方式可能是API Key也可能是Token具体看你的部署配置。星图平台一般会在控制台明确给出。模型名称这里就是InternLM2-Chat-1.8B但有时API接口可能只需要你指定路径模型名已经包含在URL里。把这些信息准备好记在一个方便的地方我们接下来写代码时会用到。假设你的API地址是https://your-deployment-url/v1/chat/completions有一个API Key是sk-xxxxxx。3. 基础API调用从第一行代码开始我们先从最简单的开始发送一个请求然后等待完整的回复。这种方式虽然不“流式”但能让我们先理解整个交互流程。3.1 构建你的第一个请求API调用本质上就是向一个特定的网址发送一段格式化的数据JSON然后接收返回的数据。对于聊天模型最常见的格式是模仿OpenAI的Chat Completions接口。下面是一个最基础的示例代码import requests import json # 你的API信息 API_URL https://your-deployment-url/v1/chat/completions API_KEY sk-xxxxxx # 请替换为你的真实Key # 准备请求头通常包含认证信息和内容类型 headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } # 准备请求体告诉模型你想聊什么 payload { model: InternLM2-Chat-1.8B, # 模型名称根据实际情况调整 messages: [ {role: user, content: 你好请介绍一下你自己。} ], max_tokens: 512, # 限制生成的最大长度 temperature: 0.7, # 控制随机性0.0最确定1.0最随机 } # 发送POST请求 response requests.post(API_URL, headersheaders, jsonpayload) # 检查请求是否成功 if response.status_code 200: # 解析返回的JSON数据 result response.json() # 提取模型回复的内容 reply result[choices][0][message][content] print(模型回复, reply) else: print(f请求失败状态码{response.status_code}) print(f错误信息{response.text})把代码里的API_URL和API_KEY换成你自己的然后运行。如果一切正常你应该能看到InternLM2-Chat-1.8B模型的自我介绍。代码要点解析headers里的Authorization是认证的关键Bearer后面跟着你的API Key。messages是一个列表里面按顺序存放对话历史。每条消息都有role角色如user或assistant和content内容。通过这个列表模型就能理解对话的上下文。max_tokens和temperature是两个常用的参数用来控制生成文本的长度和创造性。3.2 处理多轮对话聊天机器人好玩的地方在于能记住上下文。我们只需要在messages列表里依次追加对话历史就行。# 续接上面的代码假设我们已经有了第一次的回复 first_reply # 构建第二轮对话的请求 payload { model: InternLM2-Chat-1.8B, messages: [ {role: user, content: 你好请介绍一下你自己。}, {role: assistant, content: first_reply}, # 上一轮模型的回复 {role: user, content: 你刚才说的很好能再详细说说你的训练数据吗} # 用户的新问题 ], max_tokens: 512, temperature: 0.7, } # 再次发送请求 response requests.post(API_URL, headersheaders, jsonpayload) # ... 处理回复你看只要把之前的对话记录都按顺序放进messages里模型就能自然地接上话茬实现连续对话。4. 实现流式响应让文字“流”起来基础调用虽然简单但体验上差点意思。现在我们来实现更酷的流式响应。4.1 什么是流式响应普通请求就像发一封邮件你写好了寄出去然后等对方回一封完整的信。流式响应则像打电话对方一边说你一边就能听到。在技术层面开启流式响应后服务器不会等所有内容都生成完再一次性发送给你而是每生成一小段比如一个词或一句话就立刻把这一小段数据发过来。客户端收到一段就显示一段从而实现实时的打字机效果。4.2 开启流式请求实现流式响应的关键是在请求参数里加一个stream: true。同时我们不能再使用普通的requests.post()然后等待全部返回而需要处理一个持续的数据流。import requests import json API_URL https://your-deployment-url/v1/chat/completions API_KEY sk-xxxxxx headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } payload { model: InternLM2-Chat-1.8B, messages: [{role: user, content: 写一个关于人工智能的短故事大约100字。}], max_tokens: 200, temperature: 0.8, stream: True # 关键告诉服务器我们要流式输出 } # 注意这里streamTrue 参数告诉requests库我们要处理流 response requests.post(API_URL, headersheaders, jsonpayload, streamTrue) # 检查初始响应是否成功 if response.status_code ! 200: print(f请求失败: {response.status_code}) print(response.text) else: print(开始接收流式响应) collected_content # 迭代处理响应的每一行 for line in response.iter_lines(): if line: # 流式响应通常以 data: 开头 decoded_line line.decode(utf-8) if decoded_line.startswith(data: ): data_str decoded_line[6:] # 去掉 data: 前缀 if data_str [DONE]: # 流结束的标志 print(\n\n--- 生成完成 ---) break try: data json.loads(data_str) # 提取当前生成的一小段文本 delta_content data[choices][0][delta].get(content, ) if delta_content: print(delta_content, end, flushTrue) # 关键即时打印不换行 collected_content delta_content except json.JSONDecodeError: # 有时可能会收到非JSON数据如心跳包忽略即可 pass print(f\n\n完整内容\n{collected_content})运行这段代码你会看到故事是一个词一个词或一句话一句话地显示出来的而不是等了好久突然出现一整段。代码核心解析stream: True这是发给服务器的指令。requests.post(..., streamTrue)这是告诉requests库以流模式处理连接。response.iter_lines()这是一个迭代器它会持续读取从服务器发来的数据流每次返回一行。服务器发送的每一行数据通常以data:开头后面跟着一个JSON对象或[DONE]。我们从JSON对象的choices[0].delta.content路径下取出最新生成的一小段文本并立即打印出来。flushTrue确保内容立刻显示在屏幕上。4.3 构建一个更健壮的流式客户端上面的例子展示了基本原理但还不够健壮。我们把它封装成一个函数并加入更好的错误处理和状态管理。import requests import json import sys class StreamingLLMClient: def __init__(self, api_url, api_key): self.api_url api_url self.headers { Authorization: fBearer {api_key}, Content-Type: application/json } def chat_stream(self, messages, max_tokens512, temperature0.7): 发起流式聊天请求并实时打印结果 payload { model: InternLM2-Chat-1.8B, messages: messages, max_tokens: max_tokens, temperature: temperature, stream: True } try: response requests.post(self.api_url, headersself.headers, jsonpayload, streamTrue, timeout60) response.raise_for_status() # 如果状态码不是200会抛出异常 except requests.exceptions.RequestException as e: print(f\n请求发生错误: {e}) return None full_content print(模型正在思考..., end\n\n) try: for line in response.iter_lines(decode_unicodeTrue): if line: if line.startswith(data: ): data_str line[6:] if data_str [DONE]: print(\n[流结束]) break try: data json.loads(data_str) delta data.get(choices, [{}])[0].get(delta, {}) chunk delta.get(content, ) if chunk: print(chunk, end, flushTrue) full_content chunk except json.JSONDecodeError: # 忽略非JSON数据如心跳包 continue except KeyboardInterrupt: print(\n\n用户中断了生成。) return full_content except Exception as e: print(f\n处理数据流时发生错误: {e}) return full_content return full_content # 使用示例 if __name__ __main__: client StreamingLLMClient( api_urlhttps://your-deployment-url/v1/chat/completions, api_keysk-xxxxxx ) history [ {role: user, content: 用Python写一个简单的函数计算斐波那契数列的前n项。} ] reply client.chat_stream(history, max_tokens300) if reply: # 你可以把完整的回复存下来用于后续的多轮对话 history.append({role: assistant, content: reply}) print(f\n\n完整回复已保存共{len(reply)}字符。)这个类更完善一些它处理了网络请求异常、JSON解析异常还优雅地处理了用户中途按CtrlC中断的情况。decode_unicodeTrue参数让iter_lines直接返回字符串省去了手动解码的步骤。5. 错误处理与重试机制网络请求总有可能出问题服务器暂时不可用、网络波动、请求超时等等。一个健壮的客户端必须能妥善处理这些情况。5.1 常见的错误类型及处理认证错误401/403API Key错了或者没有权限。处理方式检查Key提示用户。请求格式错误400你发送的JSON数据格式不对比如少了必需的字段。处理方式检查payload结构查看错误信息。服务器错误5xx服务器端出了问题。处理方式等待一会儿重试。网络超时或连接错误网络不稳定。处理方式重试。5.2 为流式请求添加重试机制对于流式请求重试稍微复杂一点因为连接可能在中途断开。一个简单的策略是如果连接在内容生成完成前断开就重新发起请求并附带上之前已经生成好的部分内容作为上下文。下面是一个带有简单重试功能的流式对话函数import time import requests import json def robust_chat_stream_with_retry(api_url, api_key, messages, max_retries3): 带有重试机制的流式聊天函数 headers { Authorization: fBearer {api_key}, Content-Type: application/json } payload_base { model: InternLM2-Chat-1.8B, max_tokens: 512, temperature: 0.7, stream: True } full_content attempt 0 while attempt max_retries: attempt 1 print(f\n尝试第 {attempt} 次请求...) # 每次重试都使用完整的对话历史包括已生成的部分 current_messages messages.copy() if full_content: # 如果之前已经生成了一部分把它作为assistant的回复加入历史 # 这样重试时模型就知道从哪里继续 current_messages.append({role: assistant, content: full_content}) # 注意这里我们模拟用户说“继续”也可以根据场景调整提示 current_messages.append({role: user, content: 请继续。}) payload {**payload_base, messages: current_messages} try: response requests.post(api_url, headersheaders, jsonpayload, streamTrue, timeout30) response.raise_for_status() # 处理流式响应 for line in response.iter_lines(decode_unicodeTrue): if line and line.startswith(data: ): data_str line[6:] if data_str [DONE]: print(\n[生成完成]) return full_content # 成功完成返回内容 try: data json.loads(data_str) chunk data.get(choices, [{}])[0].get(delta, {}).get(content, ) if chunk: print(chunk, end, flushTrue) full_content chunk except json.JSONDecodeError: continue # 如果正常循环结束没有遇到[DONE]可能是流提前结束 print(\n[流意外结束]) # 如果已经生成了一些内容可以考虑返回 if full_content: return full_content else: # 否则准备重试 time.sleep(2) # 等待2秒后重试 continue except (requests.exceptions.Timeout, requests.exceptions.ConnectionError, requests.exceptions.ChunkedEncodingError) as e: print(f\n网络错误: {e}) if attempt max_retries: wait_time 2 ** attempt # 指数退避2, 4, 8秒... print(f等待 {wait_time} 秒后重试...) time.sleep(wait_time) else: print(已达到最大重试次数。) return full_content # 返回已生成的部分 except requests.exceptions.HTTPError as e: print(f\nHTTP错误 (状态码: {e.response.status_code}): {e.response.text[:200]}) # 如果是客户端错误4xx重试可能没用 if 400 e.response.status_code 500: return full_content else: if attempt max_retries: time.sleep(2) continue else: return full_content except Exception as e: print(f\n未知错误: {e}) return full_content return full_content # 使用示例 if __name__ __main__: API_URL https://your-deployment-url/v1/chat/completions API_KEY sk-xxxxxx conversation [ {role: user, content: 给我讲一个长篇的科幻故事开头。} ] result robust_chat_stream_with_retry(API_URL, API_KEY, conversation, max_retries3) print(f\n\n最终获取的内容长度: {len(result)} 字符)这个函数实现了几个关键点指数退避重试每次重试前等待的时间逐渐增加2秒、4秒、8秒避免对服务器造成压力。上下文恢复如果流中断前已经生成了一部分内容重试时会把这部分内容作为对话历史发送回去并提示模型“继续”从而尽量保持内容的连贯性。错误分类处理对网络错误可重试和客户端错误如400 Bad Request重试可能无用区别对待。6. 总结走完这一趟你应该已经掌握了用Python调用InternLM2-Chat-1.8B模型API的核心方法。从最基础的发送一个请求并等待回复到实现实时的流式响应看着文字一个个蹦出来再到为你的客户端穿上“防弹衣”加上错误处理和重试机制每一步都是在让这个工具变得更实用、更可靠。流式响应的体验提升是巨大的尤其是生成长文本时那种即时反馈的感觉完全不同于干等着。而良好的错误处理则能让你的程序在不太稳定的网络环境下依然保持可用性不至于因为一次短暂的波动就彻底崩溃。在实际使用中你可以根据需求继续扩展这个客户端比如加入对话历史管理、实现函数调用如果模型支持、或者做一个简单的图形界面。核心的通信逻辑已经在这里了。希望这个教程能帮你快速上手把部署好的模型真正用起来做出一些有趣或者有用的应用。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
