【AI大模型应用开发工程师特训笔记】第04讲(第7章):函数与模块
本文介绍了Python函数编程的核心概念及其在AI开发中的应用。主要内容包括:1. 函数基础:定义、调用、参数传递(位置/默认/关键字/可变参数)及返回值处理。2. 代码组织:模块化开发、变量作用域、匿名函数和标准库使用。3. 异常处理:基础语法、高级用法(多重捕获/else/finally)及自定义异常。4. 实战案例:构建健壮的LLM调用模块,集成参数校验、成本估算、重试机制等。5. 类型注解
目录
上一章我们学习了列表、字典等数据结构,能高效地组织数据。但当你写了一段有用的代码(比如调用大模型 API、计算 token 消耗),你可能会想在多个地方重复使用它。函数就是用来封装一段可重复使用的代码块,给它起个名字,随时可以“调用”。本章将学习如何定义函数、传递参数、返回值,以及如何把函数组织成模块,就像调用
openai.ChatCompletion.create()一样。所有例子都围绕大模型开发场景,为后续调用真实 API 打下基础。最后还会深入讲解异常处理,让程序更加健壮。
7.1 函数概述与定义
7.1.1 为什么需要函数
理论讲解:
在编程中,如果没有函数,当你需要重复执行某段逻辑时(如计算文本的 token 数量、校验模型参数),你不得不复制粘贴同样的代码。这样做不仅导致代码冗余、难以维护,而且一旦逻辑需要修改(例如分词规则变化),你必须修改所有复制的地方,极易出错。
函数的核心价值在于:
-
封装:将一段具有特定功能的代码打包成一个独立单元。
-
复用:可以在程序的多个位置调用同一个函数,避免重复编写。
-
可维护性:修改函数内部实现,所有调用处自动生效。
-
可读性:通过有意义的函数名(如
count_tokens),让代码逻辑一目了然。
7.1.2 函数的定义与调用
理论讲解:
Python 中使用 def 关键字定义函数。基本语法为:
def 函数名(参数1, 参数2, ...):
"""可选的文档字符串"""
函数体代码块
return 返回值 # 可选
-
函数名应遵循变量命名规则(小写字母、下划线、有意义)。
-
参数列表可以为空,也可以有多个参数。
-
函数体必须缩进(通常 4 个空格)。
-
return语句用于返回结果;若没有return,函数默认返回None。 -
调用函数时,使用
函数名(参数值)的形式。
案例代码:
# 定义函数:模拟计算token数
def count_tokens(text):
"""模拟计算文本的token数量(简单按空格分词)"""
tokens = text.split()
return len(tokens)
# 调用函数
msg = "Hello, how are you?"
num = count_tokens(msg)
print(f"消息 '{msg}' 包含 {num} 个 token")
# 输出:消息 'Hello, how are you?' 包含 4 个 token
7.1.3 函数文档字符串
理论讲解:
紧跟在函数定义之后的第一个字符串(三个双引号包裹)称为文档字符串(docstring)。它用于说明函数的作用、参数含义、返回值等。可以通过 help(函数名) 或 函数名.__doc__ 查看,也是自动生成文档的基础。
案例代码:
def count_tokens(text):
"""模拟计算文本的token数量(简单按空格分词)"""
return len(text.split())
print(count_tokens.__doc__) # 输出:模拟计算文本的token数量(简单按空格分词)
7.2 函数参数详解
7.2.1 位置参数
理论讲解:
位置参数是最常见的参数形式。调用函数时,传入的参数按照定义时的顺序依次匹配给参数。参数的个数必须与定义一致,否则会报错。
案例代码:
def send_prompt(model, prompt, temperature):
"""发送提示词到指定模型"""
print(f"使用 {model}, 温度 {temperature}, 发送: {prompt}")
# 顺序必须一致
send_prompt("gpt-4", "你好", 0.7)
# 输出:使用 gpt-4, 温度 0.7, 发送: 你好
7.2.2 默认参数
理论讲解:
可以为参数指定默认值。调用时如果不传入该参数,则使用默认值。默认参数使函数更加灵活。重要规则:默认参数必须放在所有位置参数之后。
案例代码:
def call_llm(prompt, model="gpt-3.5", temperature=0.7):
"""调用大模型(模拟)"""
print(f"模型: {model}, 温度: {temperature}")
print(f"提示词: {prompt}")
return f"这是 {model} 的回复"
# 只传必需的 prompt
response1 = call_llm("讲个笑话")
# 指定 model
response2 = call_llm("写首诗", model="gpt-4")
# 指定所有参数
response3 = call_llm("解释AI", "claude-3", 1.2)
7.2.3 关键字参数
理论讲解:
调用函数时,可以使用 参数名=值 的形式指定参数,称为关键字参数。关键字参数可以无视位置顺序,但必须写在所有位置参数之后。
案例代码:
# 使用关键字参数,顺序可以打乱
call_llm(prompt="你好", temperature=0.9, model="gpt-4")
7.2.4 可变参数:*args 和 **kwargs
理论讲解:
*args 用于接收任意多个位置参数,这些参数会被打包成一个元组。**kwargs 用于接收任意多个关键字参数,这些参数会被打包成一个字典。这在封装函数或编写装饰器时非常有用。
案例代码:
def log_api_call(func_name, *args, **kwargs):
"""记录API调用日志"""
print(f"调用函数: {func_name}")
print(f"位置参数: {args}")
print(f"关键字参数: {kwargs}")
log_api_call("chat", "gpt-4", temperature=0.8, max_tokens=100)
# 输出:
# 调用函数: chat
# 位置参数: ('gpt-4',)
# 关键字参数: {'temperature': 0.8, 'max_tokens': 100}
7.3 函数返回值
7.3.1 单个返回值
理论讲解:函数通过 return 语句返回一个值。返回后函数立即结束。
案例代码:
def get_model_cost(model):
"""返回模型每1k token的费用(美元)"""
prices = {"gpt-4": 0.03, "gpt-3.5": 0.002, "claude-3": 0.025}
return prices.get(model, 0.0)
cost = get_model_cost("gpt-4")
print(f"每1k token 费用: ${cost}")
7.3.2 多个返回值(实际返回元组)
理论讲解:Python 函数可以返回多个值,实际上这些值被封装成一个元组。调用方可以使用多个变量接收,相当于元组解包。
案例代码:
def get_model_info(model):
"""返回模型的(名称,上下文长度,费用)"""
if model == "gpt-4":
return "gpt-4", 8192, 0.03
elif model == "claude-3":
return "claude-3", 200000, 0.025
else:
return "unknown", 0, 0.0
name, limit, price = get_model_info("gpt-4")
print(f"{name}: 上下文 {limit}, 价格 {price}")
7.3.3 返回 None
理论讲解:没有显式 return 或只写 return 的函数,返回值是 None。
案例代码:
def print_warning(msg):
print(f"警告: {msg}")
# 隐式返回 None
result = print_warning("温度过高")
print(result) # 输出 None
7.4 变量作用域
理论讲解:
-
全局变量:在函数外部定义的变量,整个文件可访问。
-
局部变量:在函数内部定义的变量,只在该函数内有效。
-
函数内部可以读取全局变量,但默认不能修改(如果尝试赋值,
Python会创建新的局部变量)。 -
若要在函数内部修改全局变量,需使用
global关键字声明。但此做法不推荐,应尽量通过参数和返回值传递数据。
案例代码:
api_key = "sk-xxx" # 全局变量
total_cost = 0.0
def add_cost(amount):
global total_cost # 声明要修改全局变量
total_cost += amount
def call_api():
retries = 3 # 局部变量
print(f"使用密钥: {api_key}") # 可以读取全局变量
call_api()
add_cost(0.05)
print(total_cost) # 输出 0.05
# print(retries) # 错误!retries 在函数外不可见
7.5 匿名函数(lambda)
理论讲解:
lambda 关键字用于创建简单的、单行的匿名函数。语法为 lambda 参数: 表达式,表达式的结果就是返回值。lambda 常用于需要一个简单函数但不想用 def 正式定义的场合,例如作为 sorted、filter、map 等函数的参数。
案例代码:
# 普通函数
def square(x):
return x ** 2
# lambda 写法
square_lambda = lambda x: x ** 2
print(square(5)) # 25
print(square_lambda(5)) # 25
# AI场景:对模型列表按上下文长度排序
models = [
{"name": "gpt-4", "context": 8192},
{"name": "claude-3", "context": 200000},
{"name": "gpt-3.5", "context": 4096}
]
sorted_models = sorted(models, key=lambda m: m["context"])
for m in sorted_models:
print(m["name"], m["context"])
# 输出:
# gpt-3.5 4096
# gpt-4 8192
# claude-3 200000
运行上述代码,输出如下内容:
tianpeng@DESKTOP-4L1UF5S:~/my-ai-service$ poetry run python src/my_ai_service/loop.py
25
25
gpt-3.5 4096
gpt-4 8192
claude-3 200000
tianpeng@DESKTOP-4L1UF5S:~/my-ai-service$
7.6 模块(Module)
7.6.1 什么是模块
理论讲解:当代码越来越多时,你会想把相关的函数、变量放到一个文件里,这个文件就是一个模块(.py 文件)。通过 import 语句可以在其他文件中使用该模块中的内容。模块是组织代码的基本单位,也是实现代码复用的重要手段。
案例代码:
创建 ai_utils.py 文件:
# ai_utils.py
def count_tokens(text):
return len(text.split())
def estimate_cost(model, tokens):
prices = {"gpt-4": 0.03, "gpt-3.5": 0.002}
return tokens / 1000 * prices.get(model, 0)
MODEL_LIST = ["gpt-4", "gpt-3.5", "claude-3"]
在另一个文件中导入使用:
# 方式1:导入整个模块
import ai_utils
tokens = ai_utils.count_tokens("Hello AI")
print(tokens)
# 方式2:导入特定函数/变量
from ai_utils import count_tokens, MODEL_LIST
print(count_tokens("hi"))
# 方式3:给模块起别名
import ai_utils as ai
print(ai.estimate_cost("gpt-4", 500))
# 方式4:导入所有(不推荐,可能命名冲突)
from ai_utils import *
7.6.2 if __name__ == "__main__" 的作用
理论讲解:
当直接运行一个 .py 文件时,其 __name__ 属性会被设置为 "__main__";当该文件被作为模块导入时,__name__ 是模块名(即文件名)。因此,if __name__ == "__main__": 用于编写测试代码:只有在直接运行该脚本时才会执行,被导入时不会执行。
案例代码:
在 ai_utils.py 末尾添加:
def main():
print("测试 count_tokens:", count_tokens("hello world"))
print("测试 estimate_cost:", estimate_cost("gpt-4", 500))
if __name__ == "__main__":
main()
直接运行 python ai_utils.py 会执行测试;在其他文件中 import ai_utils 则不会。
7.6.3 Python 标准库常用模块
理论讲解:
Python 自带丰富的标准库,无需额外安装。AI 开发中常用的有:
-
json:处理 JSON 数据(API 请求/响应) -
time:时间相关(重试延迟) -
random:随机数(模拟采样) -
math:数学函数 -
requests:发送 HTTP 请求(需安装,但常与标准库一起介绍)
案例代码:
import json
import time
import random
# 模拟 API 响应
response = {
"id": "chatcmpl-123",
"choices": [{"message": {"content": "你好!"}}]
}
json_str = json.dumps(response, ensure_ascii=False)
print(json_str)
# 随机温度
temp = random.uniform(0.5, 1.5)
print(f"随机温度: {temp}")
# 延迟1秒
time.sleep(1)
运行上述代码,输出如下的内容:
tianpeng@DESKTOP-4L1UF5S:~/my-ai-service$ poetry run python src/my_ai_service/loop.py
{"id": "chatcmpl-123", "choices": [{"message": {"content": "你好!"}}]}
随机温度: 0.9758707474342543
tianpeng@DESKTOP-4L1UF5S:~/my-ai-service$
7.7 类型注解(Type Hints)
理论讲解:
类型注解是 Python 3.5+ 引入的特性,用于标注参数和返回值的类型。虽然运行时不会强制检查,但可以帮助开发者理解代码,IDE 也能提供更好的自动补全和错误提示。
案例代码:
def call_llm(prompt: str, model: str = "gpt-4", temperature: float = 0.7) -> str:
"""返回模型的回复字符串"""
return f"模拟回复: {prompt}"
# 调用时,类型错误不会在运行时阻止,但工具会有警告
result = call_llm("你好", model=123) # IDE 会提示类型不匹配
7.8 异常处理基础
7.8.1 什么是异常
理论讲解:
异常是程序执行过程中发生的错误事件。当 Python 无法继续执行时,它会“抛出”一个异常。如果没有被捕获,程序会终止并打印错误信息(traceback)。在调用大模型 API、读取配置文件、处理用户输入时,各种意外情况都可能发生(网络中断、JSON 解析失败、文件不存在等)。异常处理就是让程序在遇到错误时能够“优雅地恢复”或给出友好提示,而不是直接崩溃。
7.8.2 常见内置异常
|
异常名称 |
发生场景 |
AI 示例 |
|---|---|---|
|
|
文件不存在 |
读取不存在的 config.json |
|
|
JSON 解析失败 |
API 返回非 JSON 格式 |
|
|
字典中键不存在 |
|
|
|
列表索引越界 |
|
|
|
类型不匹配 |
字符串和整数相加 |
|
|
值不符合预期 |
|
|
|
网络连接问题 |
无法连接到 API 服务器 |
|
|
操作超时 |
API 请求超时 |
7.8.3 try-except 基础语法
理论讲解:
使用 try-except 捕获异常。基本结构:
try:
# 可能抛出异常的代码
except 异常类型:
# 发生该异常时执行的代码
案例代码:
# 没有异常处理的脆弱程序
# with open("api_key.txt", "r") as f:
# api_key = f.read().strip() # 文件不存在会崩溃
# 有异常处理的健壮程序
try:
with open("api_key.txt", "r") as f:
api_key = f.read().strip()
print(f"密钥是: {api_key}")
except FileNotFoundError:
print("错误: 找不到 api_key.txt 文件,请检查。")
api_key = "default_key"
7.9 异常处理高级用法
7.9.1 捕获多种异常
理论讲解:
一个 try 块可能抛出多种不同类型的异常,你可以分别捕获并处理。多个 except 块可以针对不同异常编写不同的处理逻辑。Exception 是几乎所有异常的基类,可以放在最后作为“兜底”,捕获其他未指定的异常。
案例代码:
def call_model(prompt, config):
try:
if config.get("bad") == True:
raise ValueError("温度设置无效")
result = {"choices": [{"message": {"content": "回复内容"}}]}
return result["choices"][0]["message"]["content"]
except KeyError:
print("错误: 返回数据结构异常,缺少必要字段")
return "默认回复"
except ValueError as e:
print(f"参数错误: {e}")
return "参数无效,无法调用"
except Exception as e:
print(f"未知错误: {e}")
return "系统错误"
print(call_model("hello", {}))
print(call_model("hello", {"bad": True}))
7.9.2 else 子句
理论讲解:
else 块中的代码只有在 try 块没有抛出任何异常时才会执行。通常用于放置那些依赖于 try 块成功执行的代码。
案例代码:
try:
with open("config.json", "r") as f:
config = json.load(f)
except FileNotFoundError:
print("配置文件不存在,使用默认配置")
config = {"model": "gpt-3.5", "temperature": 0.7}
else:
print("成功加载配置文件,并进行了额外处理")
config["loaded_at"] = "now"
7.9.3 finally 子句
理论讲解:
finally 块中的代码无论是否发生异常都会执行,非常适合用来释放资源(如关闭文件、断开网络连接)。
案例代码:
def api_call_with_cleanup():
try:
print("发起 API 调用...")
raise ConnectionError("网络中断")
except ConnectionError as e:
print(f"捕获到错误: {e}")
return "错误返回"
finally:
print("关闭连接,清理资源(一定会执行)")
result = api_call_with_cleanup()
# 输出:
# 发起 API 调用...
# 捕获到错误: 网络中断
# 关闭连接,清理资源(一定会执行)
7.9.4 主动抛出异常(raise)
理论讲解:
使用 raise 关键字可以主动抛出异常。当检测到不合理的情况时,自己抛出异常比让 Python 被动发现更清晰。
案例代码:
def set_temperature(value):
if not isinstance(value, (int, float)):
raise TypeError("温度必须是数字")
if value < 0 or value > 2:
raise ValueError("温度必须在0到2之间")
print(f"温度设置为 {value}")
try:
set_temperature(2.5) # 超出范围
except ValueError as e:
print(f"参数错误: {e}")
7.9.5 自定义异常
理论讲解:
通过继承 Exception 类可以定义自己的异常。自定义异常能够更精确地表达业务错误,便于在复杂系统中区分不同类型的错误。
案例代码:
class ModelNotSupportedError(Exception):
"""自定义异常:模型不支持"""
pass
def call_model(model_name):
supported = ["gpt-4", "claude-3"]
if model_name not in supported:
raise ModelNotSupportedError(f"模型 {model_name} 不在支持列表中")
return f"正在调用 {model_name}"
try:
call_model("llama-2")
except ModelNotSupportedError as e:
print(f"自定义异常捕获:{e}")
7.10 综合实战:健壮的 LLM 调用模块
理论讲解:
将本章所有知识整合成一个完整的模块 simple_llm.py,包含参数校验、成本估算、模拟调用、异常处理、重试机制等功能。
案例代码:
# simple_llm.py
import json
import time
import random
# 可用的模型及其配置
_MODELS = {
"gpt-4": {"context": 8192, "cost_per_1k": 0.03},
"gpt-3.5": {"context": 4096, "cost_per_1k": 0.002},
"claude-3": {"context": 200000, "cost_per_1k": 0.025},
}
# 自定义异常
class APITemporaryError(Exception):
"""临时性错误(可重试)"""
pass
class APIPermanentError(Exception):
"""永久性错误(不应重试)"""
pass
def validate_model(model: str) -> bool:
"""检查模型是否支持"""
return model in _MODELS
def get_model_context(model: str) -> int:
"""返回模型的最大上下文长度"""
return _MODELS.get(model, {}).get("context", 0)
def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
"""估算本次调用的费用(美元)"""
if model not in _MODELS:
return 0.0
price = _MODELS[model]["cost_per_1k"]
total_tokens = input_tokens + output_tokens
return total_tokens / 1000 * price
def simulate_call(prompt: str, model: str = "gpt-3.5", temperature: float = 0.7) -> dict:
"""模拟调用大模型 API"""
if not validate_model(model):
return {"error": f"不支持的模型 {model}"}
# 模拟 token 计数(粗略估计)
input_tokens = len(prompt) // 4
output_text = f"这是 {model} 对 '{prompt[:30]}...' 的模拟回复。"
output_tokens = len(output_text) // 4
cost = estimate_cost(model, input_tokens, output_tokens)
return {
"model": model,
"temperature": temperature,
"reply": output_text,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost_usd": round(cost, 6)
}
def robust_llm_call(prompt: str, max_retries: int = 3):
"""带有重试和异常处理的健壮LLM调用"""
for attempt in range(1, max_retries + 1):
try:
print(f"尝试 {attempt}/{max_retries}...")
# 模拟可能发生的各种错误
rand = random.random()
if rand < 0.1:
raise FileNotFoundError("配置文件缺失")
elif rand < 0.3:
raise ConnectionError("网络连接失败")
elif rand < 0.5:
raise json.JSONDecodeError("Expecting value", "doc", 0)
elif rand < 0.6:
raise ValueError("无效的温度参数")
# 成功情况
return f"成功回复: {prompt}"
except FileNotFoundError as e:
print(f"永久错误,无法恢复: {e}")
raise APIPermanentError("缺少必要文件,请检查配置") from e
except ValueError as e:
print(f"参数错误,无法恢复: {e}")
raise APIPermanentError("参数无效,请修改请求") from e
except (ConnectionError, json.JSONDecodeError) as e:
print(f"临时错误: {type(e).__name__}: {e}")
if attempt == max_retries:
raise APITemporaryError("重试次数用尽,请稍后再试") from e
wait = 2 ** attempt # 指数退避
print(f"等待 {wait} 秒后重试...")
time.sleep(wait)
except Exception as e:
print(f"未知错误: {e}")
raise APITemporaryError(f"未知错误: {e}") from e
return None
# 直接运行脚本时执行测试
if __name__ == "__main__":
print("测试 simple_llm 模块")
result = simulate_call("请解释什么是注意力机制", model="gpt-4", temperature=0.8)
for key, value in result.items():
print(f"{key}: {value}")
print("\n测试健壮调用(随机错误)")
for i in range(5):
print(f"\n--- 第 {i+1} 次测试 ---")
try:
res = robust_llm_call("你好,AI")
print(res)
except APIPermanentError as e:
print(f"永久性失败: {e}")
except APITemporaryError as e:
print(f"临时性失败: {e}")
time.sleep(1)
在其他脚本中使用的示例:
import simple_llm
response = simple_llm.simulate_call("写一首关于 AI 的诗", "claude-3")
print(f"回复: {response['reply']}")
print(f"消耗 token: 输入 {response['input_tokens']}, 输出 {response['output_tokens']}")
print(f"费用: ${response['cost_usd']}")
7.11 本章小结
|
概念 |
作用 |
AI 开发示例 |
|---|---|---|
|
函数定义 |
|
|
|
位置参数 |
按顺序匹配 |
|
|
默认参数 |
提供默认值 |
|
|
关键字参数 |
指定参数名调用 |
|
|
|
接收任意数量参数 |
封装函数日志 |
|
返回值 |
|
返回 token 数、费用等 |
|
作用域 |
局部/全局变量 |
|
|
lambda |
匿名单行函数 |
|
|
模块 |
|
|
|
|
区分直接运行和导入 |
模块测试代码 |
|
类型注解 |
标注参数/返回类型 |
|
|
|
捕获异常 |
文件不存在时使用默认配置 |
|
多重 |
区分不同异常 |
分别处理网络错误、超时、JSON 错误 |
|
|
无异常时执行 |
加载配置成功后打印日志 |
|
|
无论异常与否都执行 |
关闭文件或网络连接 |
|
|
主动抛出异常 |
校验参数不合法时抛出 |
|
自定义异常 |
精确表达业务错误 |
|
|
重试机制 |
处理临时性错误 |
API 限流或网络抖动时自动重试 |
更多推荐


所有评论(0)