InternLM2-Chat-1.8B模型API调用实战：Python请求库详解

InternLM2-Chat-1.8B模型API调用实战Python请求库详解刚把InternLM2-Chat-1.8B模型部署好看着服务跑起来是不是有点小激动但接下来你可能马上会问这模型怎么用起来总不能每次都打开网页去聊天吧。对于开发者来说通过代码调用API才是把AI能力集成到自己应用里的正确姿势。今天我就来聊聊怎么用Python里最常见的requests库跟这个1.8B参数的对话模型“说上话”。我会带你一步步走完整个流程从最基础的单个问题请求到能记住上下文的连续对话最后还会分享几个实际开发中容易踩的坑和解决办法。1. 准备工作环境与基础概念在开始写代码之前咱们得先把“战场”准备好。这里说的准备主要是两件事一是确保你的Python环境里有必要的工具二是理解我们要跟谁对话、怎么对话。1.1 环境检查与库安装首先打开你的终端或者命令行检查一下Python版本。InternLM2的API服务对Python 3.7及以上版本都支持良好我建议用Python 3.8或更高版本兼容性更好。python --version如果看到类似Python 3.8.10这样的输出说明版本没问题。接下来安装我们今天的主角——requests库。这个库可以说是Python里处理HTTP请求的“瑞士军刀”简单又好用。pip install requests如果你之前已经装过系统会提示Requirement already satisfied。为了确保代码清晰我们可能还会用到json库不过这个是Python自带的不需要额外安装。1.2 理解API的基本对话方式在写代码之前咱们先花两分钟搞明白InternLM2的API期待我们以什么样的格式“说话”。这就像你要给一个朋友发消息得知道是用微信还是短信是发文字还是可以发图片。InternLM2的API遵循一种现在比较通用的消息格式它期待我们发送一个JSON对象里面主要包含一个messages数组。这个数组里的每个元素都是一条消息而且有固定的角色标识role: user表示这是用户说的话也就是我们提的问题role: assistant表示这是AI助手的回答content字段里放的就是具体的文字内容比如你想问“你好介绍一下你自己”那么对应的JSON大概长这样{ messages: [ { role: user, content: 你好介绍一下你自己 } ] }模型收到这个请求后会生成回答然后以类似的JSON格式返回给我们。理解了这个基本结构写代码的时候就不会懵了。2. 第一次对话发送简单的单轮请求好了基础知识有了现在咱们来写第一个能实际跑起来的代码。我会从最简单的开始——问一个问题得到一个回答。2.1 构建你的第一个API请求假设你的InternLM2服务已经在本地的8000端口跑起来了这是默认端口那么API的地址就是http://localhost:8000/v1/chat/completions。这个/v1/chat/completions是InternLM2专门用来处理聊天补全的接口路径。我们先来写一个最基础的函数它只做一件事发送一个问题然后打印出模型的回答。import requests import json def ask_question_simple(question, api_urlhttp://localhost:8000/v1/chat/completions): 向InternLM2模型发送一个简单问题 Args: question: 要问的问题字符串 api_url: API接口地址默认是本地8000端口 Returns: 模型返回的回答文本如果出错则返回None # 1. 准备请求头 headers { Content-Type: application/json } # 2. 准备请求体就是我们要发送的数据 payload { messages: [ { role: user, content: question } ] } try: # 3. 发送POST请求 response requests.post(api_url, headersheaders, jsonpayload) # 4. 检查响应状态 if response.status_code 200: # 解析JSON响应 result response.json() # 提取AI的回答内容 answer result[choices][0][message][content] return answer else: print(f请求失败状态码{response.status_code}) print(f错误信息{response.text}) return None except requests.exceptions.RequestException as e: print(f网络请求出错{e}) return None except (KeyError, IndexError) as e: print(f解析响应数据出错{e}) return None # 试试这个函数 if __name__ __main__: question 你好请介绍一下你自己 answer ask_question_simple(question) if answer: print(模型回答) print(answer) else: print(获取回答失败)把这个代码保存为simple_chat.py然后运行它。如果一切正常你应该能看到模型的一段自我介绍。我第一次运行的时候看到它说“我是InternLM2一个由上海人工智能实验室开发的大语言模型...”那种感觉还是挺奇妙的——你的代码真的在和AI对话了。2.2 理解响应结构数据怎么来的你可能注意到了我们从响应数据里提取回答的代码有点长result[choices][0][message][content]。这看起来有点复杂咱们拆开看看每个部分是什么意思。当你收到API的响应时它其实是一个嵌套的JSON结构。我用一个实际的例子来说明# 假设response.json()返回的数据是这样的 response_data { id: chatcmpl-123456, # 这次对话的唯一ID object: chat.completion, created: 1677652288, # 时间戳 model: internlm2-chat-1.8b, # 使用的模型 choices: [ # 这是一个数组通常只有一个元素 { index: 0, message: { # 这才是AI的实际回复 role: assistant, content: 你好我是InternLM2... # 我们想要的回答在这里 }, finish_reason: stop # 生成结束的原因 } ], usage: { # 这次请求的token使用情况 prompt_tokens: 15, # 输入消耗的token数 completion_tokens: 45, # 输出消耗的token数 total_tokens: 60 # 总共消耗的token数 } } # 所以我们要拿到回答就需要 answer response_data[choices][0][message][content]理解这个结构很重要因为后续我们处理连续对话、流式输出等高级功能时都需要跟这个结构打交道。3. 让对话更智能实现连续对话单次问答虽然有用但真正的对话应该是连续的——AI能记住我们之前说过什么。比如你先问“Python是什么”接着问“它有什么优点”AI应该知道“它”指的是Python。这就是连续对话的魅力。3.1 维护对话历史记录要实现连续对话关键是要保存完整的对话历史而不仅仅是最后一条消息。每次发送请求时我们都把之前的所有对话记录一起发过去。class ChatSession: 管理一个完整的对话会话 def __init__(self, api_urlhttp://localhost:8000/v1/chat/completions): self.api_url api_url self.conversation_history [] # 保存所有的对话记录 self.headers { Content-Type: application/json } def add_message(self, role, content): 向对话历史添加一条消息 self.conversation_history.append({ role: role, content: content }) def ask(self, user_input): 向模型提问并获取回答 # 1. 先把用户的问题添加到历史记录 self.add_message(user, user_input) # 2. 准备请求数据 payload { messages: self.conversation_history } try: # 3. 发送请求 response requests.post(self.api_url, headersself.headers, jsonpayload) if response.status_code 200: result response.json() assistant_reply result[choices][0][message][content] # 4. 把AI的回答也添加到历史记录 self.add_message(assistant, assistant_reply) return assistant_reply else: print(f请求失败{response.status_code}) return None except Exception as e: print(f出错{e}) return None def clear_history(self): 清空对话历史 self.conversation_history [] def get_history(self): 获取当前的对话历史 return self.conversation_history.copy() # 使用示例 if __name__ __main__: # 创建一个对话会话 chat ChatSession() # 第一轮对话 print(你Python是什么) reply1 chat.ask(Python是什么) print(fAI{reply1}) print(\n *50 \n) # 第二轮对话 - AI应该知道“它”指的是Python print(你它有什么优点) reply2 chat.ask(它有什么优点) print(fAI{reply2}) print(\n *50 \n) # 看看完整的对话历史 print(完整的对话历史) for msg in chat.get_history(): print(f{msg[role]}: {msg[content][:50]}...) # 只显示前50个字符这个ChatSession类就像一个智能的对话管理器。它内部维护着一个对话记录列表每次你问新问题时它会把整个历史记录包括之前的所有问答一起发给模型。这样模型就能理解上下文给出连贯的回答。3.2 控制对话长度避免历史记录过长你可能会想如果对话一直进行下去历史记录不是会越来越长吗确实如此而且模型能处理的上下文长度是有限的InternLM2-Chat-1.8B通常支持4K或8K的上下文长度。当对话轮数太多时我们需要一些策略来处理过长的历史记录。最简单的方法是只保留最近N轮对话class ChatSessionWithLimit(ChatSession): 带长度限制的对话会话 def __init__(self, api_urlhttp://localhost:8000/v1/chat/completions, max_history10): super().__init__(api_url) self.max_history max_history # 最大历史记录条数 def add_message(self, role, content): 添加消息如果超过限制则移除最早的消息 super().add_message(role, content) # 如果历史记录超过限制移除最早的消息通常是成对移除 while len(self.conversation_history) self.max_history: # 通常移除最早的一对问答一个user消息和一个assistant消息 if len(self.conversation_history) 2: self.conversation_history self.conversation_history[2:] else: self.conversation_history self.conversation_history[1:] def ask(self, user_input): 提问并自动管理历史记录长度 return super().ask(user_input)这个改进版的会话类会在历史记录过长时自动清理最早的消息。max_history10意味着最多保留10条消息通常是5轮问答。你可以根据实际需要调整这个值。4. 高级功能调整生成参数有时候你可能希望AI的回答更有创意或者更保守一些有时候你可能需要更长的回答或者希望它快点结束。这些都可以通过调整生成参数来实现。4.1 常用的生成参数InternLM2的API支持多种生成参数让我介绍几个最常用的def ask_with_parameters(question, **kwargs): 带参数调整的提问函数 Args: question: 问题文本 **kwargs: 可选的生成参数如temperature、max_tokens等 Returns: 模型的回答 api_url http://localhost:8000/v1/chat/completions # 默认参数 default_params { messages: [{role: user, content: question}], temperature: 0.7, # 创造性0-2之间越高越随机 max_tokens: 512, # 最大生成长度 top_p: 0.8, # 核采样参数 stream: False # 是否流式输出 } # 用传入的参数覆盖默认值 default_params.update(kwargs) headers {Content-Type: application/json} try: response requests.post(api_url, headersheaders, jsondefault_params) if response.status_code 200: result response.json() return result[choices][0][message][content] else: print(f请求失败{response.status_code}) return None except Exception as e: print(f出错{e}) return None # 使用不同参数的例子 if __name__ __main__: question 写一个关于人工智能的短故事 print( 高创造性temperature1.2) answer1 ask_with_parameters(question, temperature1.2) print(answer1[:200] ...) # 只显示前200字符 print(\n 低创造性temperature0.3) answer2 ask_with_parameters(question, temperature0.3) print(answer2[:200] ...) print(\n 短回答max_tokens100) answer3 ask_with_parameters(question, max_tokens100) print(answer3)让我解释一下这几个参数的实际效果temperature温度这个参数控制输出的随机性。设为0时模型每次都会给出相同的回答设为较高的值如1.5时回答会更富有创意和变化。通常0.7-0.9是个不错的平衡点。max_tokens最大token数限制AI回答的最大长度。一个中文字符大约是1-2个token所以max_tokens100大概能生成50-100字。top_p核采样另一种控制随机性的方式。设置为0.8意味着模型只从概率最高的80%的词汇中选择。stream流式这个我们稍后会详细讲它让回答可以像打字一样逐个字显示。4.2 参数的实际影响对比实验为了让你更直观地理解这些参数的效果我做了个小实验def test_parameters(): 测试不同参数对回答的影响 test_question 描述一下夏天的海滩 # 测试不同temperature print(测试temperature参数) for temp in [0.1, 0.7, 1.5]: answer ask_with_parameters(test_question, temperaturetemp, max_tokens100) print(f\ntemperature{temp}) print(answer[:100] ... if len(answer) 100 else answer) # 测试不同max_tokens print(\n\n测试max_tokens参数) for tokens in [50, 150, 300]: answer ask_with_parameters(test_question, max_tokenstokens) print(f\nmax_tokens{tokens}长度{len(answer)}字) print(answer) # 测试top_p print(\n\n测试top_p参数) for top_p in [0.5, 0.8, 0.95]: answer ask_with_parameters(test_question, top_ptop_p, max_tokens100) print(f\ntop_p{top_p}) print(answer[:100] ... if len(answer) 100 else answer) if __name__ __main__: test_parameters()运行这个测试你会看到temperature0.1时回答比较保守、重复temperature1.5时回答更有创意但可能有点“跑题”max_tokens50时回答很短可能没说完max_tokens300时回答更详细完整实际使用时你可以根据场景调整这些参数。比如写创意文案时用高temperature做技术问答时用低temperature。5. 实战技巧与常见问题在实际使用API的过程中你可能会遇到各种问题。我把自己踩过的一些坑和解决办法分享给你希望能帮你少走弯路。5.1 处理网络超时和重试网络请求总有可能出问题特别是当你的应用需要稳定运行时必须考虑错误处理。import time from requests.exceptions import Timeout, ConnectionError def robust_api_call(api_url, payload, max_retries3, timeout30): 带重试机制的API调用 Args: api_url: API地址 payload: 请求数据 max_retries: 最大重试次数 timeout: 超时时间秒 Returns: 响应对象或None headers {Content-Type: application/json} for attempt in range(max_retries): try: response requests.post( api_url, headersheaders, jsonpayload, timeouttimeout ) # 检查HTTP状态码 if response.status_code 200: return response elif response.status_code 429: # 请求过多 retry_after int(response.headers.get(Retry-After, 5)) print(f请求过多{retry_after}秒后重试...) time.sleep(retry_after) continue else: print(f请求失败状态码{response.status_code}) if attempt max_retries - 1: wait_time 2 ** attempt # 指数退避 print(f{wait_time}秒后重试...) time.sleep(wait_time) except Timeout: print(f请求超时第{attempt 1}次尝试) if attempt max_retries - 1: wait_time 2 ** attempt print(f{wait_time}秒后重试...) time.sleep(wait_time) except ConnectionError: print(f连接错误第{attempt 1}次尝试) if attempt max_retries - 1: wait_time 2 ** attempt print(f{wait_time}秒后重试...) time.sleep(wait_time) except Exception as e: print(f未知错误{e}) break print(f经过{max_retries}次尝试后仍然失败) return None # 使用这个更健壮的调用函数 def ask_with_retry(question, max_retries3): 带重试的提问函数 api_url http://localhost:8000/v1/chat/completions payload { messages: [{role: user, content: question}] } response robust_api_call(api_url, payload, max_retries) if response: result response.json() return result[choices][0][message][content] else: return None这个robust_api_call函数实现了几个重要的错误处理机制指数退避重试第一次失败后等1秒第二次等2秒第三次等4秒避免给服务器造成压力429状态码处理当请求太多被限流时会读取服务器的Retry-After头按建议的时间等待超时处理设置合理的超时时间避免请求一直卡住连接错误处理网络不稳定时的重试逻辑5.2 性能优化批量处理和异步调用如果你需要同时处理多个问题或者你的应用对响应速度要求很高可以考虑下面这些优化方法。批量处理多个问题def batch_ask(questions, api_urlhttp://localhost:8000/v1/chat/completions): 批量提问注意InternLM2 API本身不支持批量这里是用循环实现的 Args: questions: 问题列表 api_url: API地址 Returns: 回答列表 answers [] for i, question in enumerate(questions, 1): print(f处理第{i}/{len(questions)}个问题...) payload { messages: [{role: user, content: question}] } try: response requests.post(api_url, jsonpayload, timeout30) if response.status_code 200: result response.json() answer result[choices][0][message][content] answers.append(answer) else: print(f问题{question[:20]}...失败状态码{response.status_code}) answers.append(None) except Exception as e: print(f问题{question[:20]}...出错{e}) answers.append(None) # 稍微延迟一下避免请求过快 time.sleep(0.5) return answers # 使用示例 if __name__ __main__: questions [ Python是什么, 如何学习Python编程, Python有哪些常用的库 ] answers batch_ask(questions) for q, a in zip(questions, answers): print(f\n问题{q}) print(f回答{a[:100]}... if a and len(a) 100 else f回答{a})使用异步请求提高效率如果你的应用支持异步import asyncio import aiohttp # 需要安装pip install aiohttp async def async_ask(question, api_urlhttp://localhost:8000/v1/chat/completions): 异步提问函数 payload { messages: [{role: user, content: question}] } async with aiohttp.ClientSession() as session: try: async with session.post(api_url, jsonpayload, timeout30) as response: if response.status 200: result await response.json() return result[choices][0][message][content] else: print(f请求失败状态码{response.status}) return None except Exception as e: print(f异步请求出错{e}) return None async def batch_async_ask(questions): 批量异步提问 tasks [async_ask(q) for q in questions] answers await asyncio.gather(*tasks, return_exceptionsTrue) # 处理异常结果 final_answers [] for ans in answers: if isinstance(ans, Exception): print(f任务出错{ans}) final_answers.append(None) else: final_answers.append(ans) return final_answers # 使用示例 if __name__ __main__: questions [ Python是什么, 如何学习Python编程, Python有哪些常用的库 ] answers asyncio.run(batch_async_ask(questions)) for q, a in zip(questions, answers): print(f\n问题{q}) if a: print(f回答{a[:100]}...)异步版本在处理大量请求时效率更高因为它可以在等待一个请求响应时处理其他请求。不过要注意异步编程的概念稍微复杂一些如果你不熟悉asyncio可以先从同步版本开始。5.3 流式输出让回答“打字”显示有时候特别是当模型生成较长回答时你可能希望看到回答逐步出现的效果就像有人在打字一样。这就是流式输出。def stream_ask(question, api_urlhttp://localhost:8000/v1/chat/completions): 流式输出提问 Args: question: 问题文本 api_url: API地址 Returns: 完整的回答文本 payload { messages: [{role: user, content: question}], stream: True, # 关键参数启用流式输出 max_tokens: 500 } headers {Content-Type: application/json} full_response try: # 使用streamTrue参数 response requests.post(api_url, headersheaders, jsonpayload, streamTrue) if response.status_code 200: print(AI, end, flushTrue) for line in response.iter_lines(): if line: line line.decode(utf-8) # 流式响应每行以data: 开头 if line.startswith(data: ): data_str line[6:] # 去掉data: if data_str [DONE]: print() # 换行 break try: data json.loads(data_str) if choices in data and len(data[choices]) 0: delta data[choices][0].get(delta, {}) if content in delta: content delta[content] print(content, end, flushTrue) full_response content except json.JSONDecodeError: continue else: print(f请求失败{response.status_code}) return None except Exception as e: print(f\n流式请求出错{e}) return None return full_response # 使用示例 if __name__ __main__: question 用一段话描述星空的美 print(f你{question}) answer stream_ask(question) print(f\n\n完整回答{answer})流式输出的关键点请求时设置stream: True服务器会返回一系列数据块而不是一次性返回完整响应每个数据块以data:开头最后以data: [DONE]结束我们需要逐块解析并显示内容这种方式的体验更好特别是生成长文本时用户不用等到完全生成完就能看到部分内容。6. 实际应用构建一个简单的聊天客户端现在我们把前面学到的所有知识整合起来构建一个简单的命令行聊天客户端。这个客户端支持连续对话、参数调整、流式输出等特性。import json import requests import readline # 用于命令行历史记录只在类Unix系统有效 class SimpleChatClient: 简单的命令行聊天客户端 def __init__(self, api_urlhttp://localhost:8000/v1/chat/completions): self.api_url api_url self.conversation_history [] self.current_params { temperature: 0.7, max_tokens: 512, stream: False } def print_help(self): 显示帮助信息 print(\n可用命令) print( /help - 显示此帮助信息) print( /clear - 清空对话历史) print( /params - 显示当前参数) print( /set 参数 值 - 设置参数如/set temperature 0.9) print( /stream on/off - 开启/关闭流式输出) print( /exit - 退出程序) print(\n支持的参数temperature(0-2), max_tokens(1-4096)) print(直接输入内容即可与AI对话) def show_params(self): 显示当前参数设置 print(\n当前参数设置) for key, value in self.current_params.items(): print(f {key}: {value}) def set_param(self, param_name, value): 设置参数 if param_name temperature: try: temp float(value) if 0 temp 2: self.current_params[temperature] temp print(ftemperature设置为{temp}) else: print(temperature必须在0-2之间) except ValueError: print(temperature必须是数字) elif param_name max_tokens: try: tokens int(value) if tokens 0: self.current_params[max_tokens] tokens print(fmax_tokens设置为{tokens}) else: print(max_tokens必须大于0) except ValueError: print(max_tokens必须是整数) else: print(f未知参数{param_name}) def toggle_stream(self, state): 切换流式输出 if state.lower() on: self.current_params[stream] True print(流式输出已开启) elif state.lower() off: self.current_params[stream] False print(流式输出已关闭) else: print(用法/stream on 或 /stream off) def ask_streaming(self, user_input): 流式提问 self.conversation_history.append({role: user, content: user_input}) payload { messages: self.conversation_history, stream: True, temperature: self.current_params[temperature], max_tokens: self.current_params[max_tokens] } headers {Content-Type: application/json} full_response try: response requests.post(self.api_url, headersheaders, jsonpayload, streamTrue) if response.status_code 200: print(AI, end, flushTrue) for line in response.iter_lines(): if line: line line.decode(utf-8) if line.startswith(data: ): data_str line[6:] if data_str [DONE]: print() break try: data json.loads(data_str) if choices in data and len(data[choices]) 0: delta data[choices][0].get(delta, {}) if content in delta: content delta[content] print(content, end, flushTrue) full_response content except json.JSONDecodeError: continue else: print(f请求失败{response.status_code}) return None except Exception as e: print(f\n请求出错{e}) return None self.conversation_history.append({role: assistant, content: full_response}) return full_response def ask_normal(self, user_input): 普通提问 self.conversation_history.append({role: user, content: user_input}) payload { messages: self.conversation_history, temperature: self.current_params[temperature], max_tokens: self.current_params[max_tokens] } headers {Content-Type: application/json} try: response requests.post(self.api_url, headersheaders, jsonpayload, timeout30) if response.status_code 200: result response.json() answer result[choices][0][message][content] print(fAI{answer}) self.conversation_history.append({role: assistant, content: answer}) return answer else: print(f请求失败{response.status_code}) return None except Exception as e: print(f请求出错{e}) return None def run(self): 运行聊天客户端 print( * 50) print(InternLM2-Chat-1.8B 聊天客户端) print(输入 /help 查看可用命令) print( * 50) while True: try: user_input input(\n你).strip() if not user_input: continue # 处理命令 if user_input.startswith(/): if user_input /help: self.print_help() elif user_input /clear: self.conversation_history [] print(对话历史已清空) elif user_input /params: self.show_params() elif user_input.startswith(/set ): parts user_input[5:].split() if len(parts) 2: self.set_param(parts[0], parts[1]) else: print(用法/set 参数 值) elif user_input.startswith(/stream ): self.toggle_stream(user_input[8:]) elif user_input /exit: print(再见) break else: print(未知命令输入 /help 查看可用命令) continue # 普通对话 if self.current_params[stream]: self.ask_streaming(user_input) else: self.ask_normal(user_input) except KeyboardInterrupt: print(\n\n检测到中断退出程序) break except EOFError: print(\n\n检测到文件结束退出程序) break except Exception as e: print(f\n发生错误{e}) if __name__ __main__: # 你可以修改这里的API地址 client SimpleChatClient(http://localhost:8000/v1/chat/completions) client.run()这个聊天客户端虽然简单但包含了我们讨论的大部分功能连续对话记住历史参数调整temperature、max_tokens流式输出开关基本的命令系统错误处理你可以直接运行它然后输入/help查看所有可用命令。试试调整temperature参数看看AI的回答风格有什么变化或者开启流式输出体验回答逐字显示的效果。7. 总结走完这一趟你应该已经掌握了用Python的requests库调用InternLM2-Chat-1.8B模型API的核心方法。从最简单的单次请求到能记住上下文的连续对话再到调整生成参数和错误处理这些技能足够你开始在自己的项目里集成AI对话能力了。实际用下来我觉得最实用的还是那个ChatSession类它把对话历史管理封装得很好用起来特别顺手。流式输出也是个不错的体验优化特别是当模型生成较长回答时用户不用干等着能看到内容逐步出现。如果你打算在生产环境使用我建议再多考虑几点一是做好错误处理和重试机制网络请求总有可能失败二是注意API的调用频率别把服务器打挂了三是根据实际场景调整生成参数创意写作可以调高temperature技术问答则应该调低。代码方面今天给的例子都是最基础的实现你可以根据自己的需求扩展。比如添加对话持久化保存到数据库或文件或者实现更复杂的上下文管理策略。最重要的是多动手试试调整参数看看效果遇到问题就查查文档或搜索一下慢慢就能掌握窍门了。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。