Python AI入门:调用大模型API实战
人工智能正在深刻改变软件开发的方式。大语言模型(LLM)的兴起让普通开发者也能在自己的项目中集成强大的AI能力。本文将从零开始,带你学会如何用Python调用大模型API,构建自己的AI应用。
一、AI与大模型概念简介
在动手写代码之前,我们先了解几个核心概念:
- 大语言模型(LLM):经过大规模文本数据训练的深度学习模型,能够理解和生成人类语言。常见的有GPT系列、Claude、GLM、Qwen等。
- API调用:大多数大模型提供商都开放了HTTP API接口,开发者可以通过发送HTTP请求来获取模型的响应,无需在本地部署庞大的模型。
- OpenAI兼容接口:许多模型提供商(如智谱AI、月之暗面、DeepSeek等)都采用了与OpenAI兼容的API格式,这意味着掌握一种调用方式,就能适配多个平台。
- Token:模型处理文本的基本单位。中文通常一个字对应1-2个token,英文一个词对应1-2个token。API调用费用通常按token数量计费。
二、环境准备
我们需要安装Python的HTTP请求库来调用API。推荐使用requests库,它是Python中最流行的HTTP库。
安装依赖
# 使用pip安装requests库
pip install requests
# 如果你使用Python 3,也可以用pip3
pip3 install requests
配置API密钥
调用大模型API需要密钥。切勿将密钥硬编码在代码中,推荐使用环境变量来管理。
# config.py - 配置文件
import os
# 从环境变量读取API密钥
API_KEY = os.getenv("AI_API_KEY", "your-api-key-here")
# API基础地址(以智谱AI为例)
BASE_URL = "https://open.bigmodel.cn/api/paas/v4"
# 模型名称
MODEL_NAME = "glm-4-flash"
# 在终端中设置环境变量(Linux/Mac)
export AI_API_KEY="your-actual-api-key"
# Windows PowerShell
$env:AI_API_KEY="your-actual-api-key"
三、调用OpenAI兼容API(对话补全)
大多数国产大模型平台都兼容OpenAI的API格式。以下是一个最基本的调用示例:
# basic_chat.py - 基础对话示例
import requests
import json
def chat_completion(prompt, system_prompt="你是一个有帮助的AI助手。"):
"""
发送对话请求到大模型API
Args:
prompt: 用户的输入文本
system_prompt: 系统提示词,定义AI的角色和行为
Returns:
模型的回复文本
"""
url = "https://open.bigmodel.cn/api/paas/v4/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {API_KEY}"
}
payload = {
"model": MODEL_NAME,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
"temperature": 0.7, # 控制回复的随机性,0-1之间
"max_tokens": 1024 # 最大生成token数量
}
try:
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status() # 检查HTTP错误
data = response.json()
reply = data["choices"][0]["message"]["content"]
return reply
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return None
# 使用示例
if __name__ == "__main__":
result = chat_completion("用Python写一个快速排序算法")
print("AI回复:", result)
代码解析:
messages是一个消息列表,每条消息包含role(角色)和content(内容)role有三种:system(系统指令)、user(用户输入)、assistant(AI回复)temperature越低,回复越确定和保守;越高,回复越有创造性max_tokens限制了回复的最大长度,避免过长响应
四、流式响应处理
普通请求需要等待模型完整生成后才能得到响应,用户体验不好。流式响应(Streaming)可以让模型边生成边返回,就像ChatGPT那样逐字显示。
# streaming_chat.py - 流式响应示例
import requests
import json
def chat_stream(prompt, system_prompt="你是一个有帮助的AI助手。"):
"""
流式调用大模型API,逐字输出结果
Args:
prompt: 用户的输入文本
system_prompt: 系统提示词
"""
url = "https://open.bigmodel.cn/api/paas/v4/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {API_KEY}"
}
payload = {
"model": MODEL_NAME,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
"temperature": 0.7,
"max_tokens": 1024,
"stream": True # 开启流式输出
}
try:
response = requests.post(
url,
headers=headers,
json=payload,
timeout=60,
stream=True # requests库的流式请求
)
response.raise_for_status()
# 逐行读取服务器返回的SSE(Server-Sent Events)数据
for line in response.iter_lines():
if line:
line = line.decode("utf-8")
# SSE格式以 "data: " 开头
if line.startswith("data: "):
data_str = line[6:] # 去掉 "data: " 前缀
# "data: [DONE]" 表示流式结束
if data_str == "[DONE]":
break
try:
data = json.loads(data_str)
content = data["choices"][0]["delta"].get("content", "")
if content:
print(content, end="", flush=True)
except json.JSONDecodeError:
continue
print() # 换行
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
# 使用示例
if __name__ == "__main__":
print("AI: ", end="", flush=True)
chat_stream("请用三句话介绍Python语言的特点")
流式响应的关键点:
- 请求体中加入
"stream": True - requests调用时设置
stream=True - 使用
response.iter_lines()逐行读取SSE数据流 - 每行数据格式为
data: {...},结束标志为data: [DONE] - 增量数据在
delta.content中(注意不是message.content)
五、多轮对话管理
真实的AI聊天需要记住上下文。实现多轮对话的核心思路是维护一个消息历史列表,每次请求时把完整的历史消息发给API。
# multi_turn_chat.py - 多轮对话管理
import requests
class ChatSession:
"""
多轮对话管理器
维护对话历史,支持上下文感知的连续对话
"""
def __init__(self, system_prompt="你是一个有帮助的AI助手。", max_history=20):
self.system_prompt = system_prompt
self.max_history = max_history # 最大保留的历史消息条数
self.messages = []
# 初始化系统消息
if system_prompt:
self.messages.append({
"role": "system",
"content": system_prompt
})
def chat(self, user_input, stream=False):
"""
发送用户消息并获取AI回复
Args:
user_input: 用户输入
stream: 是否使用流式输出
Returns:
AI的回复文本
"""
# 添加用户消息到历史
self.messages.append({
"role": "user",
"content": user_input
})
# 调用API
url = "https://open.bigmodel.cn/api/paas/v4/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {API_KEY}"
}
payload = {
"model": MODEL_NAME,
"messages": self.messages,
"temperature": 0.7,
"max_tokens": 1024,
"stream": stream
}
try:
if stream:
return self._handle_stream(url, headers, payload)
else:
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
data = response.json()
reply = data["choices"][0]["message"]["content"]
# 保存AI回复到历史
self.messages.append({
"role": "assistant",
"content": reply
})
self._trim_history()
return reply
except requests.exceptions.RequestException as e:
# 请求失败时移除刚才添加的用户消息
self.messages.pop()
print(f"请求失败: {e}")
return None
def _handle_stream(self, url, headers, payload):
"""处理流式响应"""
response = requests.post(url, headers=headers, json=payload,
timeout=60, stream=True)
response.raise_for_status()
full_reply = ""
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]":
break
try:
import json
data = json.loads(data_str)
content = data["choices"][0]["delta"].get("content", "")
if content:
print(content, end="", flush=True)
full_reply += content
except json.JSONDecodeError:
continue
print()
# 保存完整回复到历史
self.messages.append({"role": "assistant", "content": full_reply})
self._trim_history()
return full_reply
def _trim_history(self):
"""修剪历史消息,防止超出上下文长度限制"""
# 保留system消息 + 最近max_history条消息
if len(self.messages) > self.max_history + 1:
system_msg = self.messages[0]
self.messages = [system_msg] + self.messages[-self.max_history:]
def clear_history(self):
"""清除对话历史(保留系统提示)"""
self.messages = self.messages[:1]
print("对话历史已清除。")
def show_history(self):
"""显示对话历史"""
for msg in self.messages:
role = msg["role"]
content = msg["content"][:50] + "..." if len(msg["content"]) > 50 else msg["content"]
print(f"[{role}] {content}")
# 使用示例
if __name__ == "__main__":
session = ChatSession(
system_prompt="你是一个编程导师,擅长用简洁的语言解释复杂概念。",
max_history=10
)
print("=== AI编程导师(输入 quit 退出,clear 清除历史)===")
while True:
user_input = input("\n你: ").strip()
if not user_input:
continue
if user_input.lower() == "quit":
break
if user_input.lower() == "clear":
session.clear_history()
continue
print("AI: ", end="", flush=True)
session.chat(user_input, stream=True)
六、Function Calling概念
Function Calling(函数调用/工具调用)让AI模型能够调用外部函数来获取信息或执行操作。例如,AI可以调用天气查询函数来回答"今天天气怎么样"。
# function_calling.py - Function Calling示例
import requests
import json
def function_calling_demo():
"""
演示function calling:让AI调用本地函数获取当前时间
"""
# 定义可用的工具(函数)
tools = [
{
"type": "function",
"function": {
"name": "get_current_time",
"description": "获取当前的日期和时间",
"parameters": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "时区,例如 Asia/Shanghai",
"default": "Asia/Shanghai"
}
},
"required": []
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "执行数学计算",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式,例如 '2 + 3 * 4'"
}
},
"required": ["expression"]
}
}
}
]
# 本地函数实现
def get_current_time(timezone="Asia/Shanghai"):
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def calculate(expression):
try:
result = eval(expression)
return str(result)
except Exception as e:
return f"计算错误: {e}"
# 函数名到实际函数的映射
function_map = {
"get_current_time": get_current_time,
"calculate": calculate
}
url = "https://open.bigmodel.cn/api/paas/v4/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {API_KEY}"
}
user_input = "现在是几点了?再帮我算一下 123 * 456 等于多少"
payload = {
"model": MODEL_NAME,
"messages": [
{"role": "user", "content": user_input}
],
"tools": tools,
"tool_choice": "auto"
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
data = response.json()
assistant_msg = data["choices"][0]["message"]
# 检查AI是否决定调用函数
if assistant_msg.get("tool_calls"):
print("AI决定调用以下工具:")
for tool_call in assistant_msg["tool_calls"]:
func_name = tool_call["function"]["name"]
func_args = json.loads(tool_call["function"]["arguments"])
print(f" 函数: {func_name}, 参数: {func_args}")
# 调用本地函数
if func_name in function_map:
result = function_map[func_name](**func_args)
print(f" 结果: {result}")
# 将函数结果返回给AI,让它生成最终回复
payload["messages"].append(assistant_msg)
payload["messages"].append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": result
})
# 第二次请求,让AI根据函数结果生成人类可读的回复
second_response = requests.post(url, headers=headers,
json=payload, timeout=30)
second_data = second_response.json()
final_reply = second_data["choices"][0]["message"]["content"]
print(f"\n最终回复: {final_reply}")
else:
print(f"AI回复: {assistant_msg['content']}")
if __name__ == "__main__":
function_calling_demo()
七、提示工程基础
提示工程(Prompt Engineering)是优化与大模型交互效果的关键技能。好的提示词能显著提升输出质量。
提示词设计原则
- 明确具体:避免模糊表述,清楚地告诉模型你想要什么
- 提供上下文:给模型足够的背景信息
- 指定格式:明确期望的输出格式(如JSON、Markdown、表格等)
- 给出示例:few-shot prompting,通过示例教会模型预期的行为
- 分步指令:复杂任务分解为多个步骤
# prompt_engineering.py - 提示工程示例
def build_code_review_prompt(code, language="python"):
"""
构建高质量的代码审查提示词
"""
prompt = f"""请审查以下{language}代码,并从以下几个方面给出建议:
1. **代码质量**:命名是否规范、结构是否清晰
2. **潜在Bug**:是否存在逻辑错误或边界条件问题
3. **性能优化**:是否有可以改进的地方
4. **安全风险**:是否存在安全隐患
请用中文回答,每个方面给出具体建议和修改后的代码片段。
```{language}
{code}
```"""
return prompt
def build_translation_prompt(text, target_lang="英文"):
"""
构建翻译提示词,使用few-shot方式
"""
prompt = f"""请将以下中文翻译为{target_lang}。
翻译要求:
- 保持原文的语气和风格
- 专业术语使用行业标准译法
- 不要添加原文没有的内容
示例:
输入: "这个bug导致内存泄漏"
输出: "This bug causes a memory leak"
输入: "我们需要重构这个模块"
输出: "We need to refactor this module"
现在请翻译以下内容:
输入: "{text}"
输出: """
return prompt
# 使用示例
if __name__ == "__main__":
code = """
def get_data(id):
d = {}
result = db.query("select * from users where id=" + id)
for row in result:
d[row[0]] = row[1]
return d
"""
prompt = build_code_review_prompt(code)
print("代码审查提示词:")
print(prompt)
print("\n" + "="*50 + "\n")
translation_prompt = build_translation_prompt("Python是一种解释型、面向对象的编程语言")
print("翻译提示词:")
print(translation_prompt)
八、实战案例:构建一个简单的AI聊天助手
综合运用以上知识,我们来构建一个功能完整的命令行AI聊天助手。它支持多轮对话、流式输出、历史记录和多种预设角色。
# ai_assistant.py - 完整的AI聊天助手
import requests
import json
import os
from datetime import datetime
# ========== 配置 ==========
API_KEY = os.getenv("AI_API_KEY", "your-api-key-here")
BASE_URL = "https://open.bigmodel.cn/api/paas/v4/chat/completions"
MODEL = "glm-4-flash"
# ========== 预设角色 ==========
PRESETS = {
"default": {
"name": "通用助手",
"system": "你是一个有帮助的AI助手,能够回答各种问题。"
},
"coder": {
"name": "编程专家",
"system": "你是一个资深程序员,精通Python、JavaScript等多种编程语言。"
"回答代码问题时,请给出完整可运行的代码示例,并附上详细注释。"
},
"teacher": {
"name": "英语老师",
"system": "你是一位耐心的英语老师。当用户发送中文时,将其翻译为英文并解释"
"关键语法点;当用户发送英文时,纠正语法错误并给出改进建议。"
},
"writer": {
"name": "写作助手",
"system": "你是一位专业的写作助手,擅长撰写文章、润色文字、提供写作建议。"
"回复时注意文字的流畅性和逻辑性。"
}
}
class AIAssistant:
"""完整的AI聊天助手"""
def __init__(self):
self.history = []
self.system_prompt = PRESETS["default"]["system"]
self.current_preset = "default"
self.max_history = 20
def set_preset(self, name):
"""切换预设角色"""
if name in PRESETS:
self.current_preset = name
self.system_prompt = PRESETS[name]["system"]
self.history = []
print(f"\n已切换到角色: {PRESETS[name]['name']}")
else:
print(f"\n未知角色: {name}")
print(f"可用角色: {', '.join(PRESETS.keys())}")
def list_presets(self):
"""列出所有预设角色"""
print("\n可用角色:")
for key, preset in PRESETS.items():
marker = " > " if key == self.current_preset else " "
print(f"{marker}{key}: {preset['name']}")
def chat(self, user_input):
"""发送消息并获取流式响应"""
self.history.append({"role": "user", "content": user_input})
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.history[-self.max_history:])
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {API_KEY}"
}
payload = {
"model": MODEL,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2048,
"stream": True
}
try:
response = requests.post(BASE_URL, headers=headers,
json=payload, timeout=60, stream=True)
response.raise_for_status()
full_reply = ""
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]":
break
try:
data = json.loads(data_str)
content = data["choices"][0]["delta"].get("content", "")
if content:
print(content, end="", flush=True)
full_reply += content
except json.JSONDecodeError:
continue
print()
self.history.append({"role": "assistant", "content": full_reply})
self._trim_history()
except requests.exceptions.RequestException as e:
self.history.pop()
print(f"\n[错误] 请求失败: {e}")
def _trim_history(self):
if len(self.history) > self.max_history:
self.history = self.history[-self.max_history:]
def save_history(self, filename="chat_history.json"):
"""保存对话历史到文件"""
data = {
"timestamp": datetime.now().isoformat(),
"preset": self.current_preset,
"messages": self.history
}
with open(filename, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"对话历史已保存到 {filename}")
def show_help(self):
"""显示帮助信息"""
print("""
=== 命令列表 ===
/preset [名称] - 切换角色(coder/teacher/writer/default)
/presets - 列出所有角色
/save - 保存对话历史
/clear - 清除对话历史
/help - 显示帮助
/quit - 退出程序
==================
""")
def run(self):
"""运行主循环"""
print("=" * 50)
print(" AI 聊天助手")
print(f" 当前角色: {PRESETS[self.current_preset]['name']}")
print(" 输入 /help 查看命令列表")
print("=" * 50)
while True:
user_input = input("\n你: ").strip()
if not user_input:
continue
if user_input.startswith("/"):
cmd = user_input.lower()
if cmd.startswith("/preset "):
self.set_preset(cmd.split(maxsplit=1)[1].strip())
elif cmd == "/presets":
self.list_presets()
elif cmd == "/save":
self.save_history()
elif cmd == "/clear":
self.history = []
print("对话历史已清除。")
elif cmd == "/help":
self.show_help()
elif cmd in ("/quit", "/exit", "/q"):
print("再见!")
break
else:
print(f"未知命令: {user_input},输入 /help 查看帮助")
else:
print("\nAI: ", end="", flush=True)
self.chat(user_input)
if __name__ == "__main__":
assistant = AIAssistant()
assistant.run()
总结
本文带你从零开始学习了Python调用大模型API的核心知识:
- 了解了AI大模型和API调用的基本概念
- 学会了使用requests库发送HTTP请求调用API
- 掌握了流式响应的处理方式,实现逐字输出效果
- 实现了多轮对话管理,支持上下文感知
- 了解了Function Calling的概念和用法
- 学习了提示工程的基础原则和技巧
- 综合运用以上知识构建了一个完整的AI聊天助手
下一步建议:
- 尝试接入不同的模型提供商API
- 学习使用LangChain等框架构建更复杂的AI应用
- 探索RAG(检索增强生成)技术,让AI基于你的数据进行回答
- 将AI能力集成到Web应用或桌面应用中