Phi-3-mini-128k-instruct模型调用实战Python requests与异步处理详解你是不是也对那些能对话、能写代码、能回答问题的AI模型感到好奇想自己动手调用一下试试今天咱们就来聊聊怎么用Python把一个叫Phi-3-mini-128k-instruct的模型给“用”起来。别被这个名字吓到它其实就是微软推出的一个挺聪明的AI模型特点是理解能力强而且能处理很长的文本。你可能在网上看到过别人用模型生成文章或者聊天觉得挺神奇但不知道背后是怎么操作的。其实核心就是通过一种叫做API的接口用代码去“问”模型问题然后它“回答”你。这篇文章我就带你从最基础的发送一个请求开始一步步深入到怎么处理更复杂的场景比如一次问很多问题、处理超长的文档、以及让程序更稳定可靠。我会用最直白的语言和能直接运行的代码让你看完就能上手玩起来。即使你Python刚入门跟着做也完全没问题。1. 从最简单的开始你的第一个模型对话万事开头难但调用模型API的开头其实特别简单。咱们先别管那些复杂的就看看怎么让模型跟你说一句“你好”。首先你需要一个能访问这个模型的“地址”API端点和一把“钥匙”API密钥。这通常需要你在提供这个模型的云服务平台比如Azure AI Studio上创建一个资源来获取。为了演示我们假设你已经有了这些信息。import requests import json # 这是模型服务的地址和你的密钥请替换成你自己的 API_ENDPOINT “https://your-model-endpoint.cognitiveservices.azure.com/” API_KEY “your-api-key-here” # 准备好我们要问的问题 messages [ {“role”: “user”, “content”: “你好请介绍一下你自己。”} ] # 把问题包装成模型能理解的格式 payload { “messages”: messages, “max_tokens”: 500 # 限制模型回答的最大长度 } # 设置请求头告诉服务器你的密钥和内容类型 headers { “Content-Type”: “application/json”, “api-key”: API_KEY } # 最关键的一步发送请求 response requests.post(API_ENDPOINT, 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_ENDPOINT和API_KEY换成你自己的运行一下。如果一切顺利你应该能看到模型的一段自我介绍。恭喜你你已经成功完成了一次AI调用这段代码虽然简单但包含了所有核心要素构造请求内容、添加认证信息、发送请求、处理响应。无论后面的代码多复杂都是在这个基础上添砖加瓦。1.1 理解请求与响应的结构你可能对代码里的messages和返回的result结构有点疑惑我来简单解释一下。请求体我们发送的 模型通常遵循一种“对话”格式。messages是一个列表里面的每个字典代表对话中的一轮。role可以是”system”系统指令设定模型角色、”user”用户说的话、”assistant”模型之前的回答。我们第一次调用所以只放了一个用户消息。响应体模型返回的 成功时服务器会返回一个JSON。结构大致如下我们主要关心choices[0].message.content{ “id”: “某个唯一标识”, “choices”: [ { “message”: { “role”: “assistant”, “content”: “模型生成的回答在这里” }, “finish_reason”: “stop” // 结束原因 } ], “usage”: { “prompt_tokens”: 10, // 输入消耗的token数 “completion_tokens”: 50, // 输出消耗的token数 “total_tokens”: 60 // 总计 } }token你可以简单理解为“字数”的某种度量单位模型按token收费或受限制。2. 进阶技巧让对话更流畅、更高效只会问一句答一句可不够用。现实中我们可能需要连续对话或者一次性处理多个任务。这就涉及到两个重要技巧管理对话历史和异步并发请求。2.1 管理多轮对话想让模型记住之前的聊天内容你只需要在messages列表里按顺序追加历史记录就行。def chat_with_history(): # 初始化对话历史可以包含一个系统指令来设定模型行为 conversation_history [ {“role”: “system”, “content”: “你是一个乐于助人的AI助手回答要简洁明了。”} ] print(“开始与Phi-3模型对话输入‘退出’结束:”) while True: user_input input(“\n你 “) if user_input.lower() ‘退出’: print(“对话结束。”) break # 1. 将用户输入加入历史 conversation_history.append({“role”: “user”, “content”: user_input}) # 2. 发送当前完整历史给模型 payload {“messages”: conversation_history, “max_tokens”: 300} response requests.post(API_ENDPOINT, headersheaders, jsonpayload) if response.status_code 200: result response.json() assistant_reply result[‘choices’][0][‘message’][‘content’] # 3. 将模型回复也加入历史以便后续上下文连贯 conversation_history.append({“role”: “assistant”, “content”: assistant_reply}) print(f“AI {assistant_reply}”) # 可选打印本次消耗的token数做到心中有数 usage result.get(‘usage’, {}) print(f“[本次消耗token {usage.get(‘total_tokens’, ‘N/A’)}]”) else: print(f“出错啦 {response.status_code} - {response.text}”) # 简单错误处理移除刚才未成功的用户输入避免历史混乱 conversation_history.pop()运行这个函数你就可以和模型进行连续对话了。它之所以能回答“你刚才说的是...”就是因为每次请求都带上了完整的conversation_history。2.2 使用异步处理提升效率asyncio aiohttp如果你需要同时向模型问好几个不相关的问题一个一个等就太慢了。比如你想让模型分别总结三篇不同的文章。这时候异步编程就能大显身手让多个请求“同时”发出去等所有结果一起回来。同步请求是“排队”异步是“多窗口排队”。首先你需要安装异步HTTP库pip install aiohttpimport asyncio import aiohttp from typing import List async def ask_model_async(session: aiohttp.ClientSession, question: str) - str: “”“异步地向模型提问一次”“” payload { “messages”: [{“role”: “user”, “content”: question}], “max_tokens”: 200 } headers { “Content-Type”: “application/json”, “api-key”: API_KEY } try: # 注意这里用的是 session.post并且有 await async with session.post(API_ENDPOINT, jsonpayload, headersheaders) as response: if response.status 200: result await response.json() return result[‘choices’][0][‘message’][‘content’] else: error_text await response.text() return f“请求失败状态码{response.status} {error_text}” except Exception as e: return f“发生异常 {str(e)}” async def batch_ask_questions(questions: List[str]): “”“批量异步提问”“” # 创建一个共享的会话比每个请求都创建新会话高效得多 async with aiohttp.ClientSession() as session: # 为每个问题创建一个异步任务 tasks [ask_model_async(session, q) for q in questions] # 等待所有任务完成并收集结果 results await asyncio.gather(*tasks) return results # 使用示例 if __name__ “__main__”: questions [ “用一句话解释什么是机器学习。”, “Python中列表和元组的主要区别是什么”, “太阳系最大的行星是哪个” ] # 运行异步主函数 answers asyncio.run(batch_ask_questions(questions)) for q, a in zip(questions, answers): print(f“问题 {q}”) print(f“回答 {a}\n{‘-’*40}”)这段代码的核心是asyncio.gather它负责管理和并发执行所有提问任务。对于I/O密集型操作比如网络请求异步能极大提升效率。我实测过问3个问题同步方式如果每个耗时2秒总共要6秒异步方式可能总共只要2秒多一点因为它们几乎是同时等待和处理的。3. 处理复杂情况长文本、流式输出与稳定性真实项目里你不可能只发送简短的提问。可能会遇到长文档总结、代码文件分析或者希望看到模型一个字一个字地生成回答像ChatGPT那样。同时网络是不稳定的我们需要让程序更健壮。3.1 处理超长文本分块与汇总Phi-3-mini-128k-instruct 虽然支持很长的上下文128k tokens但一次请求传输大量文本可能不稳定或者响应慢。一个实用的策略是“分而治之”。思路把长文本按段落或固定长度切分成块分别让模型总结或提取关键信息最后再汇总这些结果。def split_text(text: str, chunk_size: int 2000) - List[str]: “”“按大致字符数分割文本尽量在段落末尾切断”“” chunks [] start 0 text_length len(text) while start text_length: # 计算块结束位置 end start chunk_size if end text_length: # 如果到末尾了直接取剩余部分 chunks.append(text[start:]) break # 如果不是末尾尽量在句号、换行处切断避免切断单词或句子 break_point text.rfind(‘。’, start, end) if break_point -1: break_point text.rfind(‘\n’, start, end) if break_point -1: break_point text.rfind(‘.’, start, end) if break_point -1 or break_point start: # 没找到合适的断点就在当前end处切断 break_point end chunks.append(text[start:break_point1]) start break_point 1 return chunks async def summarize_long_document_async(full_text: str): “”“异步总结长文档”“” # 1. 分块 chunks split_text(full_text, chunk_size3000) # 每块约3000字符 print(f“文档被分割成 {len(chunks)} 个块。”) # 2. 为每个块创建总结任务 summary_tasks [] async with aiohttp.ClientSession() as session: for i, chunk in enumerate(chunks): # 提示词要求模型概括本段核心内容 prompt f“请概括以下文本的主要内容要求简洁\n\n{chunk}” task ask_model_async(session, prompt) summary_tasks.append(task) # 3. 并发执行所有块的总结 chunk_summaries await asyncio.gather(*summary_tasks) # 4. 将所有块的总结拼接形成“摘要的摘要” combined_summary “\n”.join([f“第{i1}部分 {s}” for i, s in enumerate(chunk_summaries)]) final_prompt f“以下是一份长文档各个部分的概括请整合成一份完整、连贯的总体摘要\n\n{combined_summary}” # 5. 请求模型生成最终摘要 async with aiohttp.ClientSession() as session: final_summary await ask_model_async(session, final_prompt) return final_summary这个方法在处理论文、长报告时特别有用。它先化整为零再化零为整既避免了单次请求过长又能获得对整体内容的把握。3.2 实现流式输出Streaming流式输出能让用户体验更好看到文字逐个蹦出来而不是干等半天。这需要模型API支持流式响应Server-Sent Events。import json def chat_with_streaming(user_message: str): “”“演示流式响应处理”“” messages [{“role”: “user”, “content”: user_message}] payload { “messages”: messages, “max_tokens”: 500, “stream”: True # 关键参数告诉服务器开启流式输出 } response requests.post(API_ENDPOINT, headersheaders, jsonpayload, streamTrue) if response.status_code 200: print(“AI “, end“”, flushTrue) full_reply “” # 迭代流式响应的每一行 for line in response.iter_lines(): if line: line_decoded line.decode(‘utf-8’) # 流式数据通常以 “data ” 开头 if line_decoded.startswith(‘data ‘): data_str line_decoded[6] # 去掉 “data ” if data_str ‘[DONE]’: break # 流结束 try: data json.loads(data_str) # 提取增量内容 delta data[‘choices’][0].get(‘delta’, {}) content delta.get(‘content’, ‘’) if content: print(content, end“”, flushTrue) full_reply content except json.JSONDecodeError: continue print() # 换行 return full_reply else: print(f“流式请求失败 {response.status_code}”) return None注意不是所有API都支持流式传输你需要查看对应模型的文档。当streamTrue时服务器会持续发送一系列数据块每个块包含模型新生成的一小段文本直到结束。3.3 增加健壮性超时与重试网络请求总会遇到意外服务器忙、网络抖动。一个健壮的程序必须能处理这些情况。import time from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_robust_session(retries3, backoff_factor0.5): “”“创建一个带重试机制的requests会话”“” session requests.Session() # 定义重试策略 retry_strategy Retry( totalretries, # 总重试次数 backoff_factorbackoff_factor, # 重试等待时间因子 status_forcelist[429, 500, 502, 503, 504], # 遇到这些状态码才重试 allowed_methods[“POST”] # 只对POST请求重试 ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(“https//”, adapter) session.mount(“http//”, adapter) return session def robust_model_call(prompt, timeout30): “”“一个更健壮的模型调用函数包含超时和重试”“” session create_robust_session() payload {“messages”: [{“role”: “user”, “content”: prompt}], “max_tokens”: 300} try: # 设置超时连接超时和读取超时 response session.post( API_ENDPOINT, headersheaders, jsonpayload, timeouttimeout ) response.raise_for_status() # 如果状态码不是200会抛出HTTPError异常 return response.json() except requests.exceptions.Timeout: print(f“请求超时{timeout}秒。”) return None except requests.exceptions.HTTPError as e: print(f“HTTP错误 {e}”) # 可以在这里根据状态码做更细致的处理比如429请求过多等 return None except requests.exceptions.RequestException as e: print(f“请求异常 {e}”) return None finally: session.close()这个robust_model_call函数做了三件事自动重试对于临时性的服务器错误5xx或限流429会自动等待片刻后重试最多3次。超时控制如果服务器响应太慢不会一直傻等超过设定时间如30秒就放弃避免程序卡死。异常捕获把可能出现的网络错误都抓出来进行友好提示而不是让程序崩溃。4. 实战整合构建一个模型客户端类把上面所有功能组合起来封装成一个类用起来会更方便。这才是工程化的做法。import aiohttp import asyncio from typing import Optional, List, AsyncGenerator import json class Phi3MiniClient: “”“Phi-3-mini模型客户端封装类”“” def __init__(self, endpoint: str, api_key: str): self.endpoint endpoint self.headers { “Content-Type”: “application/json”, “api-key”: api_key } self.session None # 用于异步会话 def sync_chat(self, messages: List[dict], max_tokens500, temperature0.7): “”“同步聊天调用”“” payload { “messages”: messages, “max_tokens”: max_tokens, “temperature”: temperature } try: resp requests.post(self.endpoint, headersself.headers, jsonpayload, timeout30) resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: print(f“同步请求失败 {e}”) return None async def async_chat(self, messages_list: List[List[dict]]): “”“批量异步聊天”“” if not self.session: self.session aiohttp.ClientSession(headersself.headers) async def single_chat(messages): payload {“messages”: messages, “max_tokens”: 300} try: async with self.session.post(self.endpoint, jsonpayload) as resp: if resp.status 200: return await resp.json() else: return {“error”: f“Status {resp.status}”, “text”: await resp.text()} except Exception as e: return {“error”: str(e)} tasks [single_chat(msg) for msg in messages_list] results await asyncio.gather(*tasks, return_exceptionsTrue) return results async def chat_stream(self, messages: List[dict]) - AsyncGenerator[str, None]: “”“流式聊天生成器”“” payload {“messages”: messages, “max_tokens”: 500, “stream”: True} async with aiohttp.ClientSession(headersself.headers) as session: async with session.post(self.endpoint, jsonpayload) as response: if response.status 200: async for line in response.content: if line: line_decoded line.decode(‘utf-8’).strip() if line_decoded.startswith(‘data ‘): data_str line_decoded[6] if data_str ‘[DONE]’: break try: data json.loads(data_str) delta data[‘choices’][0].get(‘delta’, {}) content delta.get(‘content’, ‘’) if content: yield content except json.JSONDecodeError: continue else: yield f“[流式请求错误 {response.status}]” async def close(self): “”“关闭异步会话”“” if self.session: await self.session.close() self.session None # 使用示例 async def main(): client Phi3MiniClient(API_ENDPOINT, API_KEY) # 示例1 同步调用 print(“--- 同步调用示例 ---”) reply client.sync_chat([{“role”: “user”, “content”: “你好”}]) if reply: print(reply[‘choices’][0][‘message’][‘content’]) # 示例2 批量异步调用 print(“\n--- 批量异步调用示例 ---”) questions [ [{“role”: “user”, “content”: “什么是Python”}], [{“role”: “user”, “content”: “讲个笑话”}], ] results await client.async_chat(questions) for res in results: if isinstance(res, dict) and ‘choices’ in res: print(res[‘choices’][0][‘message’][‘content’]) # 示例3 流式输出 print(“\n--- 流式输出示例 ---”) print(“AI “, end“”, flushTrue) async for chunk in client.chat_stream([{“role”: “user”, “content”: “写一首关于春天的短诗。”}]): print(chunk, end“”, flushTrue) print() await client.close() if __name__ “__main__”: asyncio.run(main())这个Phi3MiniClient类把之前讲的功能都集成了进去提供了清晰的方法接口。同步调用用于简单场景异步批量用于提升效率流式输出用于改善交互体验。在实际项目中基于这样的类进行开发代码会整洁很多。5. 总结走完这一趟你应该对如何用Python调用像Phi-3-mini这样的AI模型有了比较全面的认识。我们从最简单的单次请求开始一步步增加了对话历史管理、异步并发处理、长文本分割、流式输出以及错误重试这些在实际开发中肯定会遇到的功能。核心其实就两点一是理解如何构造正确的HTTP请求并与模型API交互二是学会用Python的工具requests,asyncio/aiohttp来高效、稳定地完成这种交互。封装成客户端类是个好习惯能让你的代码更容易维护和复用。在实际使用中你还需要多关注API的速率限制和费用。批量异步请求虽然快但一下子发太多请求可能会被限流返回429错误这时候就需要在代码里加入适当的延迟或者使用更复杂的限流策略。另外记得处理好你的API密钥不要把它硬编码在代码里然后上传到公开的仓库用环境变量或者配置文件来管理会更安全。希望这些代码和思路能成为你探索AI应用的一块扎实的垫脚石。动手试试修改参数看看不同的temperature设置会让模型的回答有什么不同或者试试用流式输出来做一个简单的聊天界面。玩得开心获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。