当前位置：首页 > news >正文

通义千问2.5-7B-Instruct开发者指南：API调用代码实例详解

news 2026/9/5 1:09:03

通义千问2.5-7B-Instruct开发者指南：API调用代码实例详解

1. 快速了解通义千问2.5-7B-Instruct

通义千问2.5-7B-Instruct是阿里云在2024年9月发布的70亿参数指令微调模型，属于中等体量的全能型AI助手，最大的特点是完全开源且可以商用。

这个模型有几个特别实用的优势：

处理超长文本：支持128K上下文，相当于能处理百万字的长文档，写小说、分析报告都不在话下
代码能力强劲：在HumanEval测试中通过率超过85%，和340亿参数的大模型相当，日常编程辅助完全够用
数学能力突出：在MATH数据集上得分80+，超越了大多数130亿参数的模型
多语言支持：支持16种编程语言和30多种自然语言，跨语言任务也能处理
商用友好：开源协议允许商业使用，不用担心版权问题

最重要的是，它只需要4GB显存就能运行，RTX 3060这样的显卡都能流畅使用，生成速度超过每秒100个token。

2. 环境准备与API基础

2.1 安装必要的Python库

在开始调用API之前，我们需要先安装几个必要的Python库：

pip install requests jsonlines tqdm

这三个库分别用于：

requests：发送HTTP请求到API端点
jsonlines：处理JSON格式的输入输出
tqdm：显示进度条，让长时间处理更有直观反馈

2.2 设置API连接参数

假设你已经通过vLLM + Open-WebUI部署好了模型服务，通常API地址会是这样的格式：

import requests import json # API基础配置 API_URL = "http://localhost:8000/v1/completions" # 默认的vLLM API端点 API_KEY = "your-api-key-here" # 如果设置了认证 HEADERS = { "Content-Type": "application/json", "Authorization": f"Bearer {API_KEY}" # 如果需要认证 }

3. 基础API调用示例

3.1 最简单的文本生成

让我们从最简单的API调用开始，生成一段文本：

def simple_completion(prompt, max_tokens=100): """基础文本生成函数""" payload = { "model": "qwen2.5-7b-instruct", "prompt": prompt, "max_tokens": max_tokens, "temperature": 0.7, "top_p": 0.9 } try: response = requests.post(API_URL, json=payload, headers=HEADERS) response.raise_for_status() # 检查请求是否成功 result = response.json() return result['choices'][0]['text'] except Exception as e: print(f"API调用失败: {e}") return None # 使用示例 prompt = "请用Python写一个计算斐波那契数列的函数" result = simple_completion(prompt) print("生成的代码:") print(result)

3.2 带参数的进阶调用

通义千问2.5支持很多有用的参数，让我们看看如何调整生成效果：

def advanced_completion(prompt, **kwargs): """支持更多参数的生成函数""" # 默认参数 default_params = { "model": "qwen2.5-7b-instruct", "prompt": prompt, "max_tokens": 200, "temperature": 0.7, "top_p": 0.9, "frequency_penalty": 0.1, "presence_penalty": 0.1, "stop": ["\n\n", "###"], # 停止序列 "stream": False # 是否流式输出 } # 更新用户自定义参数 default_params.update(kwargs) response = requests.post(API_URL, json=default_params, headers=HEADERS) return response.json() # 使用不同参数调用 result = advanced_completion( "写一篇关于人工智能的短文", temperature=0.9, # 更高创造性 max_tokens=300, top_p=0.95 )

4. 实际应用场景代码示例

4.1 代码生成与解释

通义千问2.5的代码能力很强，我们可以用它来生成和解释代码：

def generate_code_with_explanation(task_description): """生成代码并请求解释""" prompt = f"""请为以下任务编写Python代码，并在代码后添加详细注释解释： 任务：{task_description} 要求： 1. 代码要简洁高效 2. 注释要详细易懂 3. 包含使用示例 代码：""" result = advanced_completion(prompt, max_tokens=500, temperature=0.3) return result['choices'][0]['text'] # 生成一个数据处理函数的代码 code_result = generate_code_with_explanation("读取CSV文件并计算每列的平均值") print(code_result)

4.2 长文档处理

利用128K上下文的能力，我们可以处理很长的文档：

def process_long_document(document_text, instruction): """处理长文档""" prompt = f"""请根据以下文档内容执行指令： 文档内容： {document_text} 指令：{instruction} 请确保回答基于文档内容，并尽可能详细。""" # 由于支持长上下文，我们可以直接发送长文本 result = advanced_completion(prompt, max_tokens=1000, temperature=0.1) return result['choices'][0]['text'] # 假设我们有一个长文档 long_text = "这里是一篇很长的技术文档..." # 实际使用时替换为真实长文本 summary = process_long_document(long_text, "总结文档的主要观点和技术细节")

4.3 多轮对话实现

通义千问2.5支持多轮对话，下面是一个简单的对话实现：

class ChatSession: """简单的多轮对话会话类""" def __init__(self): self.conversation_history = [] def add_message(self, role, content): """添加消息到历史""" self.conversation_history.append({"role": role, "content": content}) def get_response(self, user_message): """获取模型回复""" self.add_message("user", user_message) # 构建对话格式的prompt dialog_prompt = "\n".join( [f"{msg['role']}: {msg['content']}" for msg in self.conversation_history] ) dialog_prompt += "\nassistant: " response = advanced_completion(dialog_prompt, max_tokens=200, temperature=0.8) assistant_reply = response['choices'][0]['text'] self.add_message("assistant", assistant_reply) return assistant_reply # 使用示例 chat = ChatSession() response1 = chat.get_response("你好，请介绍Python的列表推导式") print(f"助手: {response1}") response2 = chat.get_response("能给我举个例子吗？") print(f"助手: {response2}")

5. 高级功能使用示例

5.1 工具调用（Function Calling）

通义千问2.5支持工具调用，这让它能够执行外部函数：

def handle_function_calling(user_query, available_functions): """处理工具调用请求""" prompt = f"""用户查询：{user_query} 可用工具： {json.dumps(available_functions, ensure_ascii=False, indent=2)} 请分析用户需求，如果需要调用工具，请以JSON格式输出工具名称和参数。""" response = advanced_completion( prompt, temperature=0.1, max_tokens=150 ) return response['choices'][0]['text'] # 定义可用函数 functions = [ { "name": "get_weather", "description": "获取天气信息", "parameters": { "location": "string", "date": "string" } }, { "name": "calculate_math", "description": "执行数学计算", "parameters": { "expression": "string" } } ] # 示例调用 result = handle_function_calling("北京明天天气怎么样？", functions) print("工具调用建议:", result)

5.2 JSON格式强制输出

我们可以要求模型以特定JSON格式输出，方便程序处理：

def get_structured_output(query): """获取结构化JSON输出""" prompt = f"""请根据以下查询提供结构化信息： 查询：{query} 请以以下JSON格式回复： {{ "answer": "主要回答", "explanation": "详细解释", "sources": ["相关来源1", "相关来源2"], "confidence": 0.95 }} 请确保输出是有效的JSON格式。""" response = advanced_completion( prompt, temperature=0.1, max_tokens=300 ) try: # 尝试解析JSON输出 json_output = json.loads(response['choices'][0]['text'].strip()) return json_output except json.JSONDecodeError: print("JSON解析失败，返回原始文本") return response['choices'][0]['text'] # 使用示例 structured_result = get_structured_output("解释神经网络的基本原理") print(json.dumps(structured_result, ensure_ascii=False, indent=2))

6. 批量处理与性能优化

6.1 批量请求处理

如果需要处理大量文本，可以使用批量请求提高效率：

def batch_process(prompts, batch_size=5): """批量处理多个提示""" results = [] for i in range(0, len(prompts), batch_size): batch = prompts[i:i + batch_size] batch_results = [] for prompt in batch: result = simple_completion(prompt) batch_results.append(result) results.extend(batch_results) print(f"已处理 {min(i + batch_size, len(prompts))}/{len(prompts)}") return results # 示例：批量生成产品描述 product_names = ["智能手表", "无线耳机", "笔记本电脑", "智能手机"] prompts = [f"为{product}写一段吸引人的产品描述" for product in product_names] descriptions = batch_process(prompts) for product, desc in zip(product_names, descriptions): print(f"{product}: {desc}")

6.2 流式输出处理

对于长时间生成任务，可以使用流式输出：

def stream_completion(prompt, max_tokens=200): """流式输出生成""" payload = { "model": "qwen2.5-7b-instruct", "prompt": prompt, "max_tokens": max_tokens, "temperature": 0.7, "stream": True # 启用流式输出 } response = requests.post(API_URL, json=payload, headers=HEADERS, stream=True) print("开始生成:", end=" ", flush=True) full_response = "" for line in response.iter_lines(): if line: decoded_line = line.decode('utf-8') if decoded_line.startswith('data: '): data = decoded_line[6:] # 去掉'data: '前缀 if data != '[DONE]': try: chunk = json.loads(data) token = chunk['choices'][0]['text'] print(token, end="", flush=True) full_response += token except: continue print("\n生成完成!") return full_response # 使用示例 # stream_result = stream_completion("写一个关于春天的故事")

7. 错误处理与最佳实践

7.1 健壮的API调用封装

在实际项目中，我们需要更健壮的错误处理：

def robust_api_call(prompt, max_retries=3, **kwargs): """带重试机制的API调用""" for attempt in range(max_retries): try: response = advanced_completion(prompt, **kwargs) if 'choices' in response and response['choices']: return response['choices'][0]['text'] else: raise ValueError("无效的API响应格式") except requests.exceptions.RequestException as e: print(f"网络错误 (尝试 {attempt + 1}/{max_retries}): {e}") time.sleep(2 ** attempt) # 指数退避 except ValueError as e: print(f"数据处理错误: {e}") break return None # 使用示例 result = robust_api_call( "生成一份项目计划书大纲", max_tokens=300, temperature=0.8 )

7.2 性能监控与日志

添加简单的性能监控：

import time def timed_completion(prompt, **kwargs): """带时间监控的API调用""" start_time = time.time() result = robust_api_call(prompt, **kwargs) end_time = time.time() duration = end_time - start_time if result: token_count = len(result) # 粗略估计token数量 speed = token_count / duration if duration > 0 else 0 print(f"生成完成! 耗时: {duration:.2f}秒, 速度: {speed:.1f} tokens/秒") return result # 使用示例 timed_result = timed_completion("写一段技术博客的介绍", max_tokens=150)