14. Python基础回顾：列表、字典、函数、类（适配大模型开发场景）
001、Python与大模型开发：为何基础如此重要？
昨天帮同事调试一个模型推理服务的内存泄漏问题，最终定位到一行列表操作：他在循环里反复用 += 拼接日志字符串列表，每次拼接都产生新列表对象。三小时性能调优，根源竟是基础语法细节。这件事让我重新思考——在大模型技术栈越来越复杂的今天，为什么Python基础反而更关键了？
从真实问题说起
看看这个实际遇到的例子：
# 模型推理批次处理时的日志收集
log_entries = []
for batch in data_stream:
batch_logs = process_batch(batch)
# 问题出在这行：每次循环都创建新列表
log_entries += batch_logs
# 应该用 extend()，原地扩展
内存监控显示每次循环后都有旧列表等待回收。当批次量达到十万级时，GC压力直接拖慢推理速度。这种问题不会报错，但会在生产环境慢慢暴露。
大模型开发的特殊性
现在的LLM开发，框架层抽象做得太好，反而容易让人忽视底层机制。比如你调用 transformers 库：
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("gpt-3.5-turbo")
# 下面这行，知道内部发生了什么吗？
encoded = tokenizer(texts, padding=True, truncation=True)
如果不知道 tokenizer 返回的其实是特殊字典对象（有 input_ids、attention_mask 等键），不知道 padding 参数如何影响批次矩阵形状，调试张量形状错误时就只能盲目尝试。
列表与字典：不只是容器
大模型数据处理中，列表推导式用得不对，内存立刻教你做人：
# 加载百万行训练数据
data = [json.loads(line) for line in open('dataset.jsonl')]
# 文件没关闭！而且全加载到内存
# 应该用生成器表达式
data = (json.loads(line) for line in open('dataset.jsonl'))
字典的场景更典型。模型配置管理常见这种写法：
config = {"lr": 1e-4, "batch_size": 32}
updated = config.copy() # 浅拷贝，嵌套字典还是引用
updated["optimizer"] = {"type": "Adam"}
# 这里踩过坑：如果 config 里有嵌套字典，需要 deepcopy
函数设计影响架构
很多人写工具函数时忽略参数可变性：
def preprocess(items, cache=[]):
# 灾难！默认参数是可变对象
cache.append(len(items))
return processed_items
在大模型训练流水线中，这种函数被多次调用时，cache 会累积所有历史数据。正确的做法是用 None 做默认值，函数内初始化。
更关键的是，现在流行把整个训练流程拆成小函数组合。如果每个函数都隐式依赖外部状态，调试就成了噩梦：
# 反面教材
global_step = 0
def train_step(batch):
global global_step # 别这样写
loss = model(batch)
global_step += 1
return loss
# 改进版：显式传递状态
def train_step(batch, step_counter):
loss = model(batch)
return loss, step_counter + 1
类：不仅仅是语法糖
封装模型组件时，__init__ 里放太多计算逻辑是常见误区：
class TextEmbedder:
def __init__(self, model_name):
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModel.from_pretrained(model_name) # 这里加载慢且耗内存
self.vocab_size = len(self.tokenizer) # 初始化时就计算
# 应该懒加载：用时再初始化模型
属性装饰器在大模型配置管理里特别好用：
class TrainingConfig:
def __init__(self):
self._learning_rate = None
@property
def learning_rate(self):
if self._learning_rate is None:
self._learning_rate = self._compute_lr()
return self._learning_rate
# 可以加校验逻辑
@learning_rate.setter
def learning_rate(self, value):
if not 0 < value < 1:
raise ValueError("学习率超出合理范围")
self._learning_rate = value
为什么现在更强调基础？
大模型开发有个特点：实验成本极高。一次训练可能烧掉几百GPU小时，如果因为Python对象引用错误导致中间结果异常，损失不只是时间。框架更新也快，今天用 pytorch 的 DataLoader，明天可能换 huggingface 的 Dataset，但底层的数据结构操作逻辑不变。
见过有人用错生成器，导致数据流只消费一次就空：
dataset = (x for x in range(1000))
print(sum(dataset)) # 第一次消费
print(sum(dataset)) # 第二次得到 0，生成器耗尽了
这种问题在本地测试时可能发现不了，因为测试数据量小，每次都重新创建生成器。上了生产环境处理TB级数据流，问题才暴露。
个人经验建议
-
多写纯函数：模型训练代码里，状态管理已经够复杂了。工具函数尽量做成无副作用的，输入输出明确。调试时能省一半时间。
-
掌握对象生命周期：特别是大对象（模型参数、数据集）。显式释放引用，必要时用
del加gc.collect()。在GPU内存紧张的环境里，这不是优化是必需。 -
理解浅拷贝深拷贝：模型配置嵌套字典的修改，90%的错误源于拷贝不当。
copy模块的deepcopy不是万能的，自定义对象要自己实现__deepcopy__。 -
迭代器思维：处理大模型数据流，养成惰性计算的习惯。不是所有数据都需要立刻变成列表，能用生成器的地方就用，内存友好。
-
类型提示认真写：不是给IDE看的，是给自己看的。处理复杂数据结构时，类型提示能帮你理清思路。
List[Dict[str, torch.Tensor]]这样的注释一写，数据结构就清晰了。
最后说个真实体会：去年优化过一个文本生成服务，从每秒处理10请求提升到50，没改算法没换硬件，只是把核心路径上的字典查找改成了局部变量引用，列表预分配了大小。基础语法细节，在高压力的生产环境里，会被放大成性能关键点。
大模型开发像造火箭，Python基础就是螺丝刀。工具越简单，用不好越容易出问题。下次写 for 循环时，多想一步：这个迭代对象能不能复用？这个操作会不会产生新对象？这些小习惯积累起来，就是线上服务的稳定性差异。# 002、列表（List）核心：从数据存储到Prompt工程构建
昨天review同事的代码，发现一个有意思的bug。他想把用户的历史对话记录存起来，用于后续的prompt构建，代码大概是这样的：
conversation_history = []
current_turn = ["user", "你好，介绍一下Python"]
# 错误示范 - 这里踩过坑
conversation_history.append(current_turn)
current_turn[1] = "AI: Python是一种高级编程语言..."
conversation_history.append(current_turn)
print(conversation_history) # 输出两个相同的AI回复，用户消息丢了
问题出在对列表引用的误解。在prompt工程中，这种数据污染会导致对话历史错乱，大模型收到错误的上下文，生成结果自然南辕北辙。
列表的本质：引用容器
Python的列表不存储对象本身，而是存储对象的引用。理解这一点，就理解了列表90%的陷阱。
# 正确做法：每次创建新列表
conversation_history = []
conversation_history.append(["user", "你好，介绍一下Python"])
conversation_history.append(["AI", "Python是一种高级编程语言..."])
# 或者用copy()
current_turn = ["user", "新的问题"]
conversation_history.append(current_turn.copy()) # 关键在这里
在构建大模型prompt时，我习惯用列表的列表来管理对话轮次：
def build_conversation_prompt(history, system_prompt=None):
"""构建对话prompt，history结构: [[role1, content1], [role2, content2]]"""
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
for role, content in history:
messages.append({"role": role, "content": content})
return messages
# 实际使用
chat_history = [
["user", "Python有什么特点？"],
["assistant", "简单易学、开源、丰富的库..."],
["user", "适合做什么项目？"]
]
列表推导的实战价值
教科书总教列表推导的语法，但很少说清楚什么时候用。在大模型开发中，数据清洗和转换是家常便饭。
# 从原始API响应中提取关键信息
raw_responses = [
{"id": 1, "text": "Python很好", "score": 0.9},
{"id": 2, "text": "Java也不错", "score": 0.7},
{"id": 3, "text": "C++很难", "score": 0.8}
]
# 过滤+转换一气呵成
high_quality_texts = [
item["text"] for item in raw_responses
if item["score"] > 0.75
]
# 比for循环清晰多了
但别滥用列表推导。超过两层的嵌套或者带复杂条件逻辑时，老老实实用for循环，可读性更重要。
切片操作的妙用
处理对话历史时，经常需要截取最近N轮对话，或者跳过系统消息。
# 只保留最近5轮对话（防止token超限）
full_history = [...] # 假设有20轮对话
recent_history = full_history[-5:] # 负索引从末尾开始
# 跳过前两轮（可能是旧的上下文）
trimmed_history = full_history[2:]
# 每隔一轮取一次（降采样）
sampled_history = full_history[::2]
# 反转对话顺序（某些特定场景需要）
reversed_history = full_history[::-1]
注意切片返回的是新列表，修改切片不会影响原列表，这和直接赋值引用不同。
列表作为函数参数
这是另一个容易踩坑的地方。函数内修改列表会影响外部变量，除非你明确知道自己在做什么。
def process_conversation(history, max_turns=10):
"""处理对话历史，注意：会修改传入的列表！"""
if len(history) > max_turns:
del history[max_turns:] # 直接修改原列表
return history
# 如果不想修改原列表，应该这样
def safe_process_conversation(history, max_turns=10):
history_copy = history.copy() # 或者list(history)
if len(history_copy) > max_turns:
del history_copy[max_turns:]
return history_copy
在prompt工程中，我建议保持数据不可变性，总是返回新列表，避免副作用。
性能考量
当对话历史很长时（比如处理几万条聊天记录），列表操作需要注意性能。
# 在开头插入元素很慢（O(n)）
history.insert(0, ["system", "你是一个助手"]) # 尽量避免
# 在末尾追加很快（O(1)）
history.append(["user", "新消息"])
# 频繁查找是否存在某个对话
if ["user", "具体问题"] in history: # O(n)，慢
pass
# 改用集合辅助查找（如果顺序不重要）
history_set = set(tuple(item) for item in history)
实际项目中，我常用collections.deque来管理固定长度的对话窗口，特别是需要频繁在两端操作时。
个人经验建议
-
数据不可变原则：在prompt工程中，对话历史是核心数据。每次修改都创建新列表，虽然多耗一点内存，但避免了难以调试的引用bug。内存换稳定性，值得。
-
结构一致性：定义好列表的嵌套结构后就别轻易改动。比如我固定用
[role, content]的二元组表示一轮对话，整个项目都遵守这个约定。大模型开发中，数据格式混乱是万恶之源。 -
注释写意图：别只写“这是对话列表”，要写“按时间顺序存储对话轮次，每轮是[角色, 内容]格式，用于构建ChatGPT格式的prompt”。
-
边界检查：处理列表前，先检查是否为空、长度是否超限。大模型API对token数有限制，截断逻辑要健壮。
-
工具函数封装：把列表操作封装成有业务语义的函数，比如
get_recent_turns(history, n=5)、add_user_message(history, text)。直接操作裸列表的代码，三个月后自己都看不懂。
列表看似简单，但在实际工程中，特别是大模型开发这种数据密集的场景，用对用精并不容易。掌握它的引用特性，理解每个操作的时间复杂度，才能在构建复杂prompt pipeline时游刃有余。好的数据容器选择，往往是项目代码质量的第一道分水岭。# 字典（Dict）精髓：结构化数据与大模型参数配置
昨天调试大模型推理服务时，又遇到了那个经典问题——配置文件嵌套太深，某个参数路径写错，导致整个batch推理结果异常。凌晨三点盯着日志，突然意识到：这不就是字典没玩明白的代价吗？Python字典远不止是键值对容器，它是结构化数据的骨架，更是工程实践的缩影。
从调试现场说起
上周同事提交的代码里有这么一段：
config = {
"model": "llama-3-70b",
"params": {
"temperature": 0.7,
"max_tokens": 2048
}
}
# 几百行后...
generation_config = config["params"]
generation_config["temperature"] = 0.9 # 这里动了原始配置！
问题出在引用传递上。大模型服务启动后，所有请求都用了0.9的温度参数，而测试时明明设的0.7。字典的浅拷贝坑，踩一次就够记一辈子。
字典的“三层境界”
第一层：基础容器
# 基础操作大家都懂，但注意这个细节
model_config = {
"name": "gpt-4",
"context_length": 128000,
"modalities": ["text", "vision"] # 值可以是任意类型
}
# 键不一定非得是字符串，但别玩火
weird_dict = {
42: "answer",
("layer", "norm"): "参数组", # 元组当键，需要哈希
# [1, 2]: "列表不能当键" # 这个会报错，列表不可哈希
}
实际项目中，用非字符串当键要谨慎，调试时找键值会麻烦很多。
第二层：结构化配置
大模型配置通常多层嵌套：
inference_config = {
"model": {
"name": "qwen2.5-32b",
"revision": "v1.0",
"quantization": "int4"
},
"generation": {
"strategy": "beam_search",
"beam_width": 4,
"length_penalty": 0.8
},
"hardware": {
"device": "cuda:0",
"memory_limit": "24GB"
}
}
# 安全访问多层嵌套
temperature = inference_config.get("generation", {}).get("temperature", 1.0)
# 用get链式调用，避免KeyError炸掉服务
线上服务里，每个.get()都是容错的一层保障。直接写config["a"]["b"]["c"]等于埋雷。
第三层：动态操作
# 合并配置（Python 3.9+）
base_config = {"temperature": 0.7, "top_p": 0.9}
user_override = {"temperature": 0.9, "frequency_penalty": 0.1}
final_config = base_config | user_override # 新语法，干净利落
# 老项目兼容写法
final_config = {**base_config, **user_override}
# 动态生成配置键
model_variants = ["7b", "13b", "70b"]
configs = {
f"llama3-{variant}": {
"file_path": f"./weights/llama3-{variant}.safetensors"
} for variant in model_variants
}
# 字典推导式用好了，配置文件生成能省几百行代码
大模型参数配置实战
看个真实场景——加载大模型时的参数传递：
def load_model(model_name, **kwargs):
"""加载模型，kwargs覆盖默认配置"""
defaults = {
"device_map": "auto",
"torch_dtype": "auto",
"trust_remote_code": True,
"low_cpu_mem_usage": True
}
# 用户传的参数优先
load_config = {**defaults, **kwargs}
# 特殊处理：bool型参数容易传成字符串
for key in ["trust_remote_code", "low_cpu_mem_usage"]:
if key in load_config:
if isinstance(load_config[key], str):
load_config[key] = load_config[key].lower() in ["true", "1", "yes"]
print(f"加载配置: {load_config}")
# 实际加载代码...
return None # 这里简化返回
# 调用时
load_model("Qwen2.5-7B", device_map="cuda:0", trust_remote_code="yes")
注意那个bool参数转换——很多配置系统从环境变量读值，传进来都是字符串，不处理的话"false"在Python里是True（非空字符串为真），坑就在这里。
性能与内存考量
大模型动辄几百个参数，字典设计不好内存翻倍：
# 不推荐的写法：大量小字典
layer_configs = [
{"type": "linear", "in_features": 4096, "out_features": 4096},
{"type": "layer_norm", "normalized_shape": 4096},
# ... 重复几百层，内存碎片化严重
]
# 更好的写法：结构统一化
class LayerConfig:
__slots__ = ("type", "in_features", "out_features", "normalized_shape") # 固定属性，省内存
def __init__(self, **kwargs):
for attr in self.__slots__:
setattr(self, attr, kwargs.get(attr))
# 或者用namedtuple（不可变）
from collections import namedtuple
LayerConfig = namedtuple("LayerConfig", ["type", "in_features", "out_features"])
当配置项超过50个，就该考虑用类或dataclass了。字典的灵活性是以内存和性能为代价的。
那些年踩过的坑
- 默认参数陷阱
def bad_example(config={}): # 这个默认参数是同一个字典对象！
config["processed"] = True
return config
# 连续调用会累积修改
print(bad_example()) # {'processed': True}
print(bad_example()) # {'processed': True} # 已经存在了！
- 字典顺序问题（Python 3.6之前）
# Python 3.6之前字典不保证插入顺序
# 现在没问题了，但老代码迁移时要小心
# 需要明确顺序就用OrderedDict
from collections import OrderedDict
config = OrderedDict([
("step1", "加载模型"),
("step2", "处理输入"),
("step3", "生成输出")
])
- JSON序列化限制
config = {
"model": "llama3",
"data": b"binary_data" # JSON不能序列化bytes！
"created": datetime.now() # datetime也不行
}
# 解决方案：自定义序列化
import json
from datetime import datetime
class ConfigEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, bytes):
return obj.decode("utf-8", errors="ignore")
return super().default(obj)
json_str = json.dumps(config, cls=ConfigEncoder)
个人工具箱
这些年攒下几个字典工具函数，几乎每个项目都用得到：
def deep_update(base_dict, update_dict):
"""递归更新嵌套字典"""
for key, value in update_dict.items():
if isinstance(value, dict) and key in base_dict and isinstance(base_dict[key], dict):
deep_update(base_dict[key], value)
else:
base_dict[key] = value
return base_dict
def dict_to_dot_notation(d, parent_key="", sep="."):
"""把嵌套字典展平为点分隔字符串"""
items = []
for k, v in d.items():
new_key = f"{parent_key}{sep}{k}" if parent_key else k
if isinstance(v, dict):
items.extend(dict_to_dot_notation(v, new_key, sep=sep).items())
else:
items.append((new_key, v))
return dict(items)
def safe_get(d, path, default=None):
"""安全获取嵌套字典值，支持点分隔路径"""
keys = path.split(".")
current = d
for key in keys:
if isinstance(current, dict) and key in current:
current = current[key]
else:
return default
return current
写在最后
字典用得好不好，关键看能不能区分“什么时候用字典，什么时候不用”。我的经验法则是：如果键是固定的、结构是稳定的、需要类型提示的，用dataclass；如果键是动态的、结构变化频繁、需要灵活增删的，用字典。
大模型配置管理有个实用技巧：用两层结构。第一层用dataclass定义配置骨架，第二层用字典存动态参数。这样既享受了类型安全，又保留了灵活性。
最后提醒一句：生产环境里，所有配置字典都应该有版本字段。模型参数、预处理逻辑、甚至字典结构本身都会变，没有版本号的话，三个月后根本不知道当时用的什么配置。这是用无数个不眠之夜换来的教训。
技术债迟早要还，但好的数据结构能让还款期限延长几年。字典不是万能的，但理解它的精髓后，你会发现很多复杂问题不过是嵌套字典的展开与折叠。# 004、函数（Function）设计：模块化思维与Agent功能封装
上周排查一个对话系统的bug，发现有个状态判断逻辑在三个不同的地方重复出现，每处实现都有细微差异。某个边缘场景下，三处判断结果竟然不一致，导致Agent行为出现分裂。这种问题在早期代码中很常见——功能直接铺开写，看似快速，后期维护时四处打补丁，最终变成一团乱麻。
从铺开到封装
先看个典型例子，早期版本中处理用户查询的代码可能是这样的：
# 原始版本：功能直接铺开
user_input = "帮我查一下北京明天天气"
if "天气" in user_input and "北京" in user_input:
# 调用天气API
city = "北京"
date = "明天"
# 这里踩过坑：日期格式转换容易出错
if date == "明天":
target_date = get_tomorrow_date()
# 调用API、解析结果、格式化输出...
# 同样的逻辑在别处又写了一遍
几周后，类似的逻辑出现在查询航班、查询股价的代码里，每个地方都重新实现了一遍日期解析、城市提取。某天API格式变化，需要修改日期处理逻辑，就得在代码里搜出所有相关位置——漏掉一处就出bug。
函数：第一层抽象
把重复逻辑抽成函数，是最直接的改进：
def extract_location_and_date(text):
"""从文本提取地点和时间
别这样写：函数做了两件不相关的事，后期难复用
"""
# 混合处理逻辑...
return location, date
# 更好的做法：单一职责
def extract_location(text):
"""只负责提取地点"""
# 使用更精准的匹配逻辑
for city in CITY_LIST:
if city in text:
return city
return None
def parse_date_mention(text):
"""只负责解析时间提及"""
# 这里可以独立优化时间解析算法
if "明天" in text:
return get_tomorrow_date()
if "下周" in text:
return get_next_week_date()
return datetime.now()
函数设计的关键在于“单一职责”——一个函数只做好一件事。这听起来简单，实际编码时很容易违反。我习惯在写完后问自己：这个函数名能否准确描述其功能？如果名字里出现“和”、“与”、“同时”这类词，通常意味着职责过多。
参数设计的实战经验
参数设计直接影响函数复用性。看个实际场景：
# 初版：参数硬编码
def call_weather_api():
"""调用天气API"""
url = "https://api.weather.com/v1"
headers = {"Authorization": "Bearer xxx"} # 硬编码密钥，大忌！
# ...
# 改进版：配置参数化
def call_weather_api(api_key, city, date=None, units="metric"):
"""可配置的天气查询
date为None时查当前天气
units支持metric/imperial
"""
# 参数验证前置，避免传到API层才报错
if not api_key or not city:
raise ValueError("必要参数缺失")
# 默认值处理
date = date or datetime.now().strftime("%Y-%m-%d")
# 构建请求
params = {
"city": city,
"date": date,
"units": units,
"apikey": api_key # 这里注意参数命名一致性
}
# ...
参数设计时我遵循几个原则：必需参数放前面，可选参数放后面；参数名要自解释；布尔型参数尽量少用（容易迷惑）；超过5个参数时考虑用字典或数据类封装。
返回值的处理艺术
函数返回什么，怎么返回，直接影响调用方体验：
# 反例：返回类型不一致
def process_user_query(query):
"""处理用户查询"""
if not query:
return None # 有时返回None
if "天气" in query:
return {"weather": data} # 有时返回字典
return "抱歉，不理解您的请求" # 有时返回字符串
# 调用方每次都得判断类型，极易出错
# 正例：返回统一结构
def process_user_query_v2(query):
"""返回标准化的Response对象"""
if not query:
return Response(success=False, error="空查询")
try:
# 核心处理逻辑
result = analyze_query(query)
return Response(success=True, data=result)
except Exception as e:
# 异常捕获在函数内部处理
logger.error(f"处理失败: {e}")
return Response(success=False, error=str(e))
# 配合数据类更清晰
from dataclasses import dataclass
@dataclass
class Response:
success: bool
data: Any = None
error: str = ""
在Agent开发中，我倾向于让函数返回明确的结构化数据，而不是混合类型。这样上层调用代码更健壮，错误处理也更一致。
函数作为构建块：组装Agent能力
真正的威力在于用函数组装复杂能力。假设我们要构建一个旅行助手Agent：
# 基础能力函数
def detect_intent(text):
"""意图识别"""
# 使用模型或规则判断
return intent
def extract_entities(text, intent):
"""实体抽取，根据意图调整策略"""
# ...
def validate_constraints(entities):
"""约束条件验证"""
# 检查时间合理性、地点有效性等
return validation_result
# 组合成完整流程
def travel_assistant_pipeline(user_input):
"""旅行助手处理流水线"""
# 每一步都是独立的函数，可单独测试
intent = detect_intent(user_input)
entities = extract_entities(user_input, intent)
# 验证环节
validation = validate_constraints(entities)
if not validation.valid:
return format_error_response(validation)
# 根据意图路由到不同处理器
if intent == "flight_search":
return handle_flight_search(entities)
elif intent == "hotel_booking":
return handle_hotel_booking(entities)
# ...
这种模块化设计让Agent能力像乐高积木一样可组合。要新增一个“租车”功能，只需实现对应的处理函数，然后在流水线中添加路由分支。
闭包与装饰器：高级封装技巧
在Agent开发中，经常需要给函数添加额外能力，比如日志、重试、缓存：
def with_retry(max_attempts=3):
"""重试装饰器
注意：要处理幂等性问题，不是所有函数都适合重试
"""
def decorator(func):
def wrapper(*args, **kwargs):
last_error = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
last_error = e
if attempt < max_attempts - 1:
wait_time = 2 ** attempt # 指数退避
time.sleep(wait_time)
raise last_error
return wrapper
return decorator
# 使用装饰器增强API调用
@with_retry(max_attempts=3)
def call_external_api(url, params):
"""调用外部API，自动重试"""
# 原始实现
response = requests.get(url, params=params)
response.raise_for_status()
return response.json()
装饰器让横切关注点（日志、重试、缓存）与核心逻辑分离，代码更干净。但要注意装饰器会增加调用栈深度，调试时可能要多跳一层。
个人经验与建议
函数设计水平直接决定代码质量。多年踩坑后，我形成了几个习惯：
-
先写调用代码。在实现函数前，先想好它应该怎么被调用。从调用者角度思考接口设计，往往能得到更简洁的API。
-
函数长度是重要信号。如果一个函数超过50行，大概率做了太多事。屏幕一屏显示不下的函数，基本都需要拆分。
-
参数传递避免“隧道效应”。不要为了底层函数的一个参数，让中间所有函数都传递它。考虑用上下文对象或配置类封装。
-
异常处理要一致。要么在函数内部完全处理异常，返回错误码；要么让异常抛出，由上层统一处理。最糟的是部分处理部分抛出。
-
为函数写文档字符串，哪怕只有一行。三个月后回头看代码，你会感谢自己。特别是参数说明和返回类型，能节省大量调试时间。
-
Agent场景特别要注意：功能函数尽量保持纯净，避免副作用。那些需要访问外部状态、调用API的函数，单独标记和管理。这样测试时能轻松mock，系统也更稳定。
最后记住，函数不是越复杂越好。最好的函数往往简单到看起来“显然应该这样实现”。把复杂问题拆解成简单函数的组合，这才是模块化思维的核心。下次写代码时，不妨先停下来想想：这段逻辑未来会在别处用到吗？如果是，现在就把它封装起来。# 005、类（Class）入门：面向对象与LLM工具类抽象
昨天调试一个对话历史管理功能，代码里散落着七八个列表和字典，每次新增字段都得改三四个函数。半夜两点盯着屏幕突然清醒：这堆数据明明在描述同一个逻辑实体，为什么不封装成类？今天我们就聊聊Python中的类——不止是语法，更要看它在大模型开发场景下如何简化工程结构。
从混乱字典到清晰对象
先看个真实案例。早期做LLM对话历史管理时，我写过这样的代码：
# 别这样写！后期维护会疯掉
conversation_history = []
current_turn = {
"user_input": "",
"llm_response": "",
"timestamp": None,
"tokens_used": 0,
"cost": 0.0
}
def add_turn(history, user_msg, llm_msg):
turn = current_turn.copy() # 这里踩过坑：直接赋值会导致引用问题
turn["user_input"] = user_msg
turn["llm_response"] = llm_msg
turn["timestamp"] = datetime.now()
# 后面还有五六个字段要设置...
history.append(turn)
两周后需求变更，要增加情绪分析和上下文标记。我在五个不同文件里找到了操作这个字典的代码，改到第三处时已经想重写了。
类的救赎
同样的功能用类来实现：
class DialogueTurn:
"""单轮对话的完整上下文"""
def __init__(self, user_input: str, llm_response: str):
self.user_input = user_input
self.llm_response = llm_response
self.timestamp = datetime.now()
self.tokens_used = 0
self.cost = 0.0
# 新增字段只需在这里加一次
self.sentiment_score = 0.0
self.needs_followup = False
def calculate_cost(self, price_per_token: float):
"""成本计算逻辑封装在对象内部"""
self.cost = self.tokens_used * price_per_token
return self.cost
def to_api_format(self):
"""转换格式的方法也放在这里"""
return {
"role": "user",
"content": self.user_input
}
现在添加新字段只需要改一个地方，所有相关操作自然聚集在类内部。编辑器还能提供代码补全，不用再查文档确认字典键名。
为什么大模型开发特别需要类
LLM应用开发有个特点：数据对象复杂但模式统一。比如提示词模板、对话历史、工具调用结果，这些都可以抽象成类。
看个工具调用封装的例子：
class LLMTool:
"""所有LLM工具的基类"""
def __init__(self, name: str, description: str):
self.name = name
self.description = description
self.call_count = 0 # 自动统计调用次数
self.last_call_time = None
def get_schema(self):
"""生成OpenAI兼容的工具描述"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description
}
}
def execute(self, **kwargs):
"""子类必须实现的具体逻辑"""
raise NotImplementedError
class CalculatorTool(LLMTool):
"""计算器工具的具体实现"""
def __init__(self):
super().__init__(
name="calculator",
description="执行数学计算"
)
def execute(self, expression: str) -> float:
self.call_count += 1
self.last_call_time = datetime.now()
# 这里可以加安全检查、日志记录等
try:
result = eval(expression) # 生产环境要用更安全的计算库
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {str(e)}"
这种结构下，新增工具只需要继承LLMTool，团队协作时接口完全统一。监控、日志、错误处理都在基类里搞定。
类变量和实例变量的坑
刚用类时容易混淆这两个概念：
class ConfigManager:
# 类变量：所有实例共享
api_endpoint = "https://api.openai.com/v1"
def __init__(self, api_key: str):
# 实例变量：每个对象独立
self.api_key = api_key
self.request_count = 0
# 这里有个经典陷阱
config1 = ConfigManager("key1")
config2 = ConfigManager("key2")
# 修改类变量会影响所有实例
ConfigManager.api_endpoint = "https://custom.endpoint"
print(config1.api_endpoint) # 也变了！
print(config2.api_endpoint) # 也变了！
# 而实例变量各自独立
config1.request_count = 100
print(config2.request_count) # 还是0
大模型配置管理时，我建议用实例变量存储连接参数，用类变量存储常量（比如默认超时时间）。
私有属性的实用技巧
Python没有真正的私有变量，但约定用下划线表示“别直接访问”：
class APIClient:
def __init__(self, api_key: str):
self._api_key = api_key # 单下划线：提示这是内部变量
self._cache = {} # 外部可以访问，但不建议
def _make_request(self, prompt: str):
"""私有方法：外部不应该直接调用"""
# 这里可能有重试逻辑、token计算等
pass
@property
def api_key(self):
"""通过property安全访问"""
return f"{self._api_key[:8]}..." # 只显示前8位
@api_key.setter
def api_key(self, value: str):
"""设置时可以做验证"""
if not value.startswith("sk-"):
raise ValueError("Invalid API key format")
self._api_key = value
这种封装在团队协作时特别有用。上周实习生差点把API密钥打印到日志里，幸好用了property做了脱敏。
继承与组合的选择
很多教程一上来就讲继承，但实际工程中组合往往更实用：
# 继承方式：有时候会陷入深层次结构
class AdvancedLLMClient(BasicLLMClient, CacheMixin, LoggingMixin):
# 多重继承容易混乱
pass
# 组合方式：更清晰灵活
class LLMAgent:
def __init__(self):
self.llm_client = OpenAIClient()
self.cache = LRUCache()
self.logger = StructuredLogger()
def chat(self, message: str):
# 明确调用各个组件
cached = self.cache.get(message)
if cached:
self.logger.log_cache_hit()
return cached
response = self.llm_client.chat(message)
self.cache.set(message, response)
return response
大模型应用经常要集成多个服务（向量数据库、缓存、监控），组合模式让每个组件保持独立，测试和替换都更方便。
个人经验建议
-
不要过度设计：开始写代码时用字典和函数没问题，当同一组数据被三个以上函数操作时，就该考虑封装成类了。
-
类名要像名词，方法名要像动词：
DialogueTurn、PromptTemplate这种命名，新人看一眼就知道是什么。calculate_cost()比cost_calc()更自然。 -
先写
__init__方法：确定对象需要哪些数据才能正常工作，这些就是实例变量。其他方法都是围绕这些数据展开的操作。 -
在大模型项目中建立工具类库：把
PromptTemplate、MessageHistory、TokenCounter这些常用组件类化，项目规模扩大后复用性会体现出来。 -
多用
@dataclass简化样板代码：Python 3.7+的dataclass可以自动生成__init__等方法，适合纯数据类。
面向对象不是银弹，但在管理复杂状态时确实能减少认知负担。下次写LLM应用时，试着把散落各处的提示词模板、对话历史、API配置封装成类，你会发现调试和扩展都轻松不少。好的类设计就像给代码分文件夹，东西多了才知道有多重要。# 006、列表推导式与生成器：高效数据处理与流式响应
上周排查一个线上服务的内存泄漏问题，发现某段数据处理代码在加载一个300MB的JSON文件时，内存直接飙到2GB。打开源码一看，满屏的for循环嵌套临时列表，数据在内存里来回倒腾。这让我想起很多刚接触Python的工程师容易忽略的两个利器：列表推导式和生成器。今天咱们就聊聊怎么用它们写出既高效又优雅的代码。
从内存爆炸的代码说起
先看那段出问题的代码片段：
# 原始代码 - 别这样写！
results = []
with open('large_data.json') as f:
data = json.load(f) # 这里踩过坑：一次性加载整个大文件
for item in data:
if item['status'] == 'active':
processed = heavy_process(item) # 假设这是个耗时操作
results.append(processed)
return results
问题很明显：json.load()一次性把整个文件读到内存，后面又创建了results列表存储所有结果。当文件很大时，内存直接撑爆。
列表推导式：紧凑但需谨慎
列表推导式是Python的语法糖，能把多行循环压缩成一行：
# 改进版 - 简洁多了
active_items = [item for item in data if item['status'] == 'active']
results = [heavy_process(item) for item in active_items]
看起来清爽，但本质上还是在内存中创建了两个完整列表。对于小数据集没问题，但遇到大数据时依然危险。
真正的问题是：我们真的需要所有数据同时存在内存里吗？ 很多时候我们只是需要逐个处理数据项，这时候就该生成器上场了。
生成器：懒加载的艺术
生成器的核心思想是“按需计算”，用()代替[]：
# 生成器版本 - 内存友好
active_items = (item for item in data if item['status'] == 'active')
results = (heavy_process(item) for item in active_items)
注意这里results不是列表，而是一个生成器对象。它不会立即执行heavy_process，只有当你迭代它时才会逐个处理。
生成器函数：处理复杂逻辑
当处理逻辑复杂时，可以用yield定义生成器函数：
def process_stream(file_path):
"""流式处理JSON文件，避免内存爆炸"""
with open(file_path, 'r') as f:
# 使用ijson库流式解析，避免一次性加载
for item in ijson.items(f, 'item'):
if item['status'] == 'active':
yield heavy_process(item)
# 使用方式
for result in process_stream('large_data.json'):
send_to_api(result) # 处理一个发送一个
这种模式特别适合大模型开发场景。比如处理百万条对话数据时，你可以边读边处理边训练，而不是等所有数据加载完才开始。
生成器表达式在API响应中的应用
在大模型服务中，经常需要处理流式响应。假设我们要批量处理用户查询：
def batch_process_queries(queries):
"""模拟大模型批量处理"""
for query in queries:
# 模拟模型推理耗时
time.sleep(0.1)
yield {
'query': query,
'response': f"Processed: {query[:10]}..."
}
# 客户端可以这样接收流式响应
async def handle_client_stream(websocket):
queries = await websocket.receive_json()
# 关键在这里：立即开始返回第一个结果
for result in batch_process_queries(queries):
await websocket.send_json(result)
# 不用等所有查询处理完，处理完一个就发一个
这种模式让客户端能尽快收到第一个响应，用户体验上感觉更快，服务端内存压力也小。
性能对比实测
写个简单测试感受下差异：
import memory_profiler
@memory_profiler.profile
def test_list_comprehension():
data = range(10**6)
squared = [x**2 for x in data] # 立即创建百万个元素的列表
return sum(squared)
@memory_profiler.profile
def test_generator():
data = range(10**6)
squared = (x**2 for x in data) # 只是个生成器对象
return sum(squared) # sum()会驱动生成器逐个计算
跑一下就会发现，列表推导式版本的内存峰值明显更高，因为它在内存中构建了完整的列表。而生成器版本几乎保持恒定内存。
几个容易踩的坑
- 生成器只能消费一次：
gen = (x for x in range(3))
list(gen) # [0, 1, 2]
list(gen) # [] 第二次就是空的！
-
调试时注意：生成器不执行时不会暴露错误。有时候你以为代码没问题，实际迭代时才发现异常。
-
不要过度使用：简单的
for循环其实更易读。如果逻辑复杂，别硬塞进一行推导式。
个人经验建议
在实际项目中，我通常这样选择：
- 数据量小且需要重复访问 → 用列表推导式，缓存结果
- 数据量大或只需单次遍历 → 用生成器表达式，节省内存
- 处理网络流或文件流 → 用生成器函数，配合
yield控制流程 - 需要链式处理 → 生成器管道，每个
yield像流水线的一个环节
最近做AI服务开发时，生成器成了我的首选。大模型动辄处理几十万条数据，用生成器配合流式API，内存占用能降一个数量级。特别是处理实时对话时，用户说完第一句就能开始生成响应，不用等整段话说完。
记住一个原则：数据就像水流，能流动就别囤积。生成器让数据流动起来，这才是处理大规模数据的正确姿势。
下次看到内存使用异常增长时，先检查下是不是该用生成器的地方用了列表。这个简单的替换，往往能解决大部分内存问题。# 字典进阶操作：JSON序列化与大模型I/O实战笔记
昨天调试大模型API时又遇到一个典型问题：从外部服务拿到一段JSON响应，直接json.loads()转成字典后，某个字段明明存在却报KeyError。这种问题在对接大模型服务时太常见了——响应结构复杂、嵌套深、字段可能缺失，直接裸用字典就像在雷区散步。今天咱们就聊聊字典在JSON和大模型场景下的那些实战技巧。
从踩坑案例说起
# 模拟大模型API返回的复杂结构
api_response = """
{
"choices": [
{
"message": {
"role": "assistant",
"content": "Python中字典的JSON序列化..."
},
"finish_reason": "stop"
}
],
"usage": {
"prompt_tokens": 25,
"completion_tokens": 42
}
}
"""
import json
data = json.loads(api_response)
# 新手常写的危险代码
try:
content = data['choices'][0]['message']['content']
print(f"模型回复：{content}")
except KeyError as e:
print(f"字段不存在：{e}")
问题来了：如果API某次返回的choices数组为空，或者message字段缺失，这段代码直接崩溃。大模型服务的响应结构虽然文档上有，但实际运行中总有意外。
安全访问的几种姿势
方案一：get()方法链式调用
# 这样写安全多了，但有点啰嗦
content = data.get('choices', [{}])[0].get('message', {}).get('content')
if content is None:
content = "默认回复"
方案二：try-except局部化
# 我个人更偏好这种，意图明确
try:
content = data['choices'][0]['message']['content']
except (KeyError, IndexError):
content = None
方案三：封装工具函数
def safe_get(d, *keys, default=None):
"""链式安全获取嵌套字典值"""
current = d
for key in keys:
if isinstance(current, dict) and key in current:
current = current[key]
else:
return default
return current
# 用起来很清爽
content = safe_get(data, 'choices', 0, 'message', 'content', default='')
tokens = safe_get(data, 'usage', 'completion_tokens', default=0)
JSON序列化的那些坑
大模型场景下经常要把字典序列化成JSON发给API，这里有几个细节要注意：
config = {
'model': 'gpt-4',
'temperature': 0.7,
'max_tokens': 1000,
'stream': True,
# 注意这个datetime对象！
'created_at': datetime.now(),
# 还有这个自定义类实例
'user_context': UserContext(id=123)
}
# 直接序列化会报错：Object of type datetime is not JSON serializable
# json_str = json.dumps(config) # 报错！
# 正确做法：自定义序列化器
def custom_serializer(obj):
"""处理常见非JSON类型"""
if isinstance(obj, datetime):
return obj.isoformat() # 转成ISO格式字符串
elif hasattr(obj, '__dict__'):
return obj.__dict__ # 简单对象转字典
elif isinstance(obj, set):
return list(obj) # 集合转列表
raise TypeError(f"无法序列化类型：{type(obj)}")
json_str = json.dumps(config, default=custom_serializer, ensure_ascii=False)
实际项目中，我习惯把常用配置封装成类：
class ModelRequest:
def __init__(self, model="gpt-3.5-turbo"):
self.model = model
self.messages = []
self.temperature = 0.7
def add_message(self, role, content):
self.messages.append({"role": role, "content": content})
def to_json(self):
# 控制哪些字段需要序列化
return json.dumps({
"model": self.model,
"messages": self.messages,
"temperature": self.temperature
}, ensure_ascii=False)
@classmethod
def from_json(cls, json_str):
# 反序列化时重建对象
data = json.loads(json_str)
req = cls(model=data.get('model'))
req.messages = data.get('messages', [])
return req
大模型I/O中的字典技巧
1. 消息历史管理
大模型对话需要维护消息历史，用字典列表比元组列表更易扩展：
# 不好的写法：用元组存储，难以扩展字段
history = [("user", "你好"), ("assistant", "你好！")]
# 好的写法：字典列表，随时可以加字段
history = [
{"role": "user", "content": "你好", "timestamp": "2024-01-01T10:00:00"},
{"role": "assistant", "content": "你好！", "tokens": 5}
]
# 添加系统提示也很自然
history.insert(0, {"role": "system", "content": "你是一个Python专家"})
2. 流式响应处理
大模型的流式响应需要边接收边解析：
def process_stream_response(stream):
"""处理流式JSON响应"""
buffer = ""
for chunk in stream:
buffer += chunk.decode('utf-8')
# 尝试解析完整的JSON对象
while True:
try:
# 这里有个技巧：用json.JSONDecoder的raw_decode
# 可以解析字符串开头的第一个完整JSON对象
decoder = json.JSONDecoder()
obj, idx = decoder.raw_decode(buffer)
# 成功解析，处理数据
yield obj
# 移除已处理的部分
buffer = buffer[idx:].lstrip()
except json.JSONDecodeError:
# 数据不完整，继续接收
break
3. 配置管理
大模型参数多，用字典管理比一堆参数更清晰：
# 硬编码参数，难以维护
def call_model(prompt, temp=0.7, top_p=0.9, max_tokens=100, ...):
pass
# 用字典管理，配置和代码分离
model_config = {
"default": {
"temperature": 0.7,
"max_tokens": 2000,
"top_p": 0.9
},
"creative": {
"temperature": 1.2,
"max_tokens": 1000
},
"precise": {
"temperature": 0.2,
"max_tokens": 500
}
}
def call_model(prompt, config_name="default", **overrides):
config = model_config[config_name].copy()
config.update(overrides) # 允许临时覆盖
# 使用config字典调用API
性能考虑
处理大模型返回的大字典时要注意内存：
# 如果只需要部分字段，尽早过滤
big_response = get_huge_model_response() # 假设返回几十MB的JSON
# 不好的做法：全量解析后再取字段
# data = json.loads(big_response) # 可能内存爆炸
# 好的做法：流式解析或选择性解析
import ijson # 第三方库，流式解析大JSON
with open('big_response.json', 'rb') as f:
# 只解析需要的部分
for item in ijson.items(f, 'choices.item.message.content'):
process_content(item)
个人经验建议
-
防御性编程：处理外部API返回的字典时，永远假设字段可能缺失、类型可能不对。大模型服务还在快速迭代，今天有的字段明天可能就变了。
-
序列化保持兼容：在字典里加新字段时，考虑旧代码的反序列化。我习惯给每个配置字典加个
version字段，方便后续做迁移。 -
不要过度设计：早期项目直接用字典，等模式稳定了再封装成类。见过有人一开始就设计复杂的ORM式映射，结果需求一变全得重写。
-
善用Python 3.8+特性：
:=运算符在处理嵌套字典时能简化代码，dataclasses可以替代简单的配置字典。 -
日志记录完整字典：调试时把整个请求/响应字典记到日志，但记得先脱敏。用
json.dumps(indent=2)让输出更易读。
大模型开发本质上是数据流处理，字典作为Python的核心数据结构，用好了能让代码既健壮又灵活。关键是想清楚数据从哪里来、到哪里去、中间可能怎么变——这比记住多少API语法都重要。# 函数装饰器与闭包：增强LLM交互与日志追踪
昨天排查一个对话历史丢失的问题，发现是LLM接口调用时上下文管理函数被意外覆盖了。这种问题在多层函数嵌套时特别隐蔽，最终用装饰器加闭包解决了。今天咱们就聊聊Python里这个既实用又容易踩坑的特性。
从实际场景说起
假设你正在封装一个LLM调用函数，需要记录每次请求的耗时和token使用量。最直接的做法是在每个函数里写日志代码：
def call_llm(prompt):
start_time = time.time()
# 原始调用逻辑
result = real_llm_call(prompt)
end_time = time.time()
print(f"耗时: {end_time - start_time:.2f}秒")
return result
但如果有十几个这样的函数呢？复制粘贴的代码很快就会失控。这时候就该装饰器上场了。
装饰器的本质
装饰器说白了就是个“函数包装器”。它接收一个函数，返回一个新函数，中间可以插入各种额外逻辑。看个最简单的例子：
def log_time(func):
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs) # 执行原函数
print(f"{func.__name__} 耗时: {time.time() - start:.2f}s")
return result
return wrapper
@log_time
def query_llm(prompt):
# 真实的LLM调用
return "模拟响应"
那个@log_time语法糖，等价于query_llm = log_time(query_llm)。调用query_llm()时，实际执行的是wrapper()函数。
闭包为什么重要
注意上面wrapper函数里引用了外部函数的func参数。这种内部函数记住外部变量（即使外部函数已返回）的现象，就是闭包。
闭包在装饰器里至关重要，它让wrapper能记住要装饰的是哪个函数。没有闭包，装饰器就失去了灵魂。我遇到过有人把装饰器写成类属性，结果所有实例共享同一个闭包变量，调试了半天才发现问题。
带参数的装饰器
实际项目中，我们可能需要更灵活的装饰器。比如根据日志级别决定是否输出：
def log_with_level(level="INFO"):
def decorator(func):
def wrapper(*args, **kwargs):
if level == "DEBUG":
print(f"[DEBUG] 调用 {func.__name__}")
result = func(*args, **kwargs)
return result
return wrapper
return decorator
@log_with_level(level="DEBUG")
def debug_call():
pass
这里实际上是三层嵌套：最外层接收参数，中间层接收函数，最内层才是包装逻辑。第一次写的时候容易晕，多写几次就习惯了。
保留函数元信息
直接使用装饰器有个坑：原函数的名字、文档字符串等元信息会被覆盖。比如：
@log_time
def get_embedding(text):
"""获取文本嵌入向量"""
pass
print(get_embedding.__name__) # 输出 'wrapper'，不是 'get_embedding'
解决方法是使用functools.wraps：
from functools import wraps
def log_time(func):
@wraps(func) # 这行是关键
def wrapper(*args, **kwargs):
# ... 包装逻辑
pass
return wrapper
这个细节很容易忽略，等用反射或者看文档时才发现问题。
在LLM开发中的实际应用
结合咱们的大模型开发场景，装饰器能做的事情很多。分享几个我实际在用的：
请求重试装饰器：
def retry_on_failure(max_retries=3):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise
print(f"第{attempt+1}次重试，错误: {e}")
time.sleep(2 ** attempt) # 指数退避
return None
return wrapper
return decorator
上下文管理装饰器：
def maintain_context(max_turns=10):
def decorator(func):
@wraps(func)
def wrapper(self, *args, **kwargs):
# 自动维护对话轮次
if len(self.conversation_history) >= max_turns:
self.conversation_history.pop(0)
return func(self, *args, **kwargs)
return wrapper
return decorator
性能监控装饰器：
def monitor_performance():
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
# 这里可以接入监控系统
if elapsed > 1.0: # 超过1秒警告
print(f"警告: {func.__name__} 执行过慢: {elapsed:.2f}s")
return result
return wrapper
return decorator
闭包的陷阱
闭包虽好，但有个经典坑。看这段代码：
def create_handlers():
handlers = []
for i in range(3):
def handler():
return i * 2
handlers.append(handler)
return handlers
for handler in create_handlers():
print(handler()) # 输出什么？
三个handler都会输出4，而不是预期的0, 2, 4。因为闭包记住的是变量i本身，不是创建时的值。等执行时，循环早已结束，i的值是2。
修正方法是用默认参数捕获当前值：
def handler(i=i): # 这里！默认参数在定义时求值
return i * 2
这个坑我在异步回调里踩过，调试了整整一个下午。
装饰器的调试技巧
装饰器增加了调用栈深度，出错时堆栈跟踪会变复杂。我的经验是：
- 在装饰器内部打印
func.__name__，确认装饰的是哪个函数 - 复杂装饰器先用简单函数测试，再应用到业务代码
- 使用
inspect.signature检查参数签名是否被意外修改 - 装饰器内部异常要重新抛出，不要静默吞掉
个人建议
装饰器是Python的利器，但在LLM开发这种多层架构中要谨慎使用。我现在的原则是：装饰器只负责横切关注点（日志、监控、重试），不涉及核心业务逻辑。如果一个装饰器超过50行，就该考虑拆成辅助函数或类了。
另外，类装饰器（__call__方法）在某些场景下更清晰，特别是需要维护状态的装饰逻辑。但普通函数装饰器在可读性上通常更胜一筹。
最后提醒一点：装饰器在导入模块时就会执行，而不是在函数调用时。如果装饰器里有耗时的初始化操作，考虑改成惰性初始化，避免拖慢程序启动速度。
好的装饰器设计能让代码干净不少，但过度使用也会让调用链变得难以追踪。掌握这个度，需要在实际项目中多踩几个坑才能找到感觉。# 009、类的继承与多态：构建可扩展的LLM应用框架
从一次深夜调试说起
上周在封装一个多模型调用框架时，我写了这样的代码：
class ChatHandler:
def __init__(self, model_name):
self.model_name = model_name
def generate(self, prompt):
if self.model_name == "gpt-4":
return self._call_openai(prompt)
elif self.model_name == "claude":
return self._call_anthropic(prompt)
elif self.model_name == "llama":
return self._call_local(prompt)
# 每加一个模型就要改这里...
凌晨两点，当我准备接入第七个模型时，看着满屏的if-elif突然清醒了——这根本不是面向对象，这是面向if-else编程。第二天我重构了整个设计，核心就是今天要聊的继承与多态。
继承：别重复造轮子
先看个实际场景。假设我们要处理不同LLM的API调用，每个模型都有共同点（都需要API key，都要处理超时），也有差异点（端点URL不同，参数格式不同）。
糟糕的做法（我最初写的）：
class GPT4Client:
def __init__(self, api_key):
self.api_key = api_key
self.endpoint = "https://api.openai.com/v1/chat/completions"
def call(self, prompt):
headers = {"Authorization": f"Bearer {self.api_key}"}
# GPT-4专用逻辑...
class ClaudeClient:
def __init__(self, api_key):
self.api_key = api_key
self.endpoint = "https://api.anthropic.com/v1/messages"
def call(self, prompt):
headers = {"x-api-key": self.api_key}
# Claude专用逻辑...
发现没？api_key的初始化重复了，超时处理逻辑也会重复，日志记录也会重复… 每加一个模型就得重写一遍。
用继承重构后：
class BaseLLMClient:
"""所有LLM客户端的基类，把通用逻辑放这里"""
def __init__(self, api_key, timeout=30):
self.api_key = api_key
self.timeout = timeout
self.logger = self._setup_logger() # 日志初始化也统一处理
def _setup_logger(self):
# 统一的日志配置
import logging
return logging.getLogger(self.__class__.__name__)
def validate_prompt(self, prompt):
"""通用的prompt验证逻辑"""
if not prompt or len(prompt.strip()) == 0:
raise ValueError("Prompt不能为空")
return prompt.strip()
def call(self, prompt):
# 这里故意不实现，让子类必须自己实现
raise NotImplementedError("子类必须实现call方法")
class GPT4Client(BaseLLMClient):
def __init__(self, api_key, **kwargs):
# 先调用父类的初始化
super().__init__(api_key, **kwargs)
self.endpoint = "https://api.openai.com/v1/chat/completions"
# GPT-4特有的初始化...
def call(self, prompt):
self.validate_prompt(prompt) # 复用父类方法
self.logger.info(f"调用GPT-4: {prompt[:50]}...")
# GPT-4特有的调用逻辑
import openai
# ... 具体实现
class ClaudeClient(BaseLLMClient):
def __init__(self, api_key, **kwargs):
super().__init__(api_key, **kwargs)
self.endpoint = "https://api.anthropic.com/v1/messages"
def call(self, prompt):
self.validate_prompt(prompt)
self.logger.info(f"调用Claude: {prompt[:50]}...")
# Claude特有的调用逻辑
# ... 具体实现
现在要加新模型，只需要继承BaseLLMClient，实现自己的call方法就行。通用逻辑全在父类里，这就是继承的核心价值——代码复用和层次化设计。
多态：写一次，适配所有
多态听起来抽象，其实很简单：用同样的接口调用不同的对象。在LLM应用里，这太有用了。
class ModelRouter:
"""模型路由——多态的典型应用"""
def __init__(self):
self.clients = {} # 存放不同模型的客户端实例
def register(self, model_name, client_instance):
"""注册模型客户端"""
# 这里不关心client_instance的具体类型
# 只要它有call方法就行（鸭子类型）
self.clients[model_name] = client_instance
def generate(self, model_name, prompt):
client = self.clients.get(model_name)
if not client:
raise ValueError(f"未注册的模型: {model_name}")
# 关键在这里：不管client是GPT4Client还是ClaudeClient
# 我都用同样的方式调用
return client.call(prompt)
def batch_generate(self, prompts, model_selector=None):
"""批量生成，可以动态选择模型"""
results = []
for i, prompt in enumerate(prompts):
if model_selector:
# 根据策略选择模型（比如简单prompt用便宜模型）
model_name = model_selector(prompt, i)
else:
model_name = "default"
# 多态的魔力：这里不需要知道具体是哪个模型
result = self.generate(model_name, prompt)
results.append(result)
return results
# 使用示例
router = ModelRouter()
router.register("gpt-4", GPT4Client(api_key="sk-..."))
router.register("claude", ClaudeClient(api_key="claude-key..."))
router.register("llama", LlamaClient(model_path="./llama-model"))
# 看这里！同样的generate调用，背后是不同的模型
result1 = router.generate("gpt-4", "解释量子计算")
result2 = router.generate("claude", "写一首关于Python的诗")
多态让我们的代码面向接口而非实现。ModelRouter不关心具体模型怎么实现，只关心它们都有call方法。这带来的扩展性是巨大的——明天要加Google的Gemini，只需要写个GeminiClient继承BaseLLMClient，然后注册到router里就行，其他代码一行都不用改。
实际踩坑：继承的陷阱
继承用不好会写出更糟糕的代码。分享几个我踩过的坑：
1. 过度继承（ inheritance overkill）
# 别这样写！这是“继承滥用”
class AdvancedGPT4Client(GPT4Client):
"""加了缓存功能的GPT-4客户端"""
def __init__(self, api_key, cache_size=100):
super().__init__(api_key)
self.cache = {}
self.cache_size = cache_size
def call(self, prompt):
if prompt in self.cache:
return self.cache[prompt]
result = super().call(prompt)
self._add_to_cache(prompt, result)
return result
问题在哪？缓存应该是横切关注点，用组合而不是继承：
class CachedLLMClient:
"""用组合实现缓存，更灵活"""
def __init__(self, client_instance, cache_size=100):
self.client = client_instance # 组合一个客户端实例
self.cache = {}
self.cache_size = cache_size
def call(self, prompt):
if prompt in self.cache:
print("[缓存命中]")
return self.cache[prompt]
result = self.client.call(prompt)
self._add_to_cache(prompt, result)
return result
# 这样用
gpt_client = GPT4Client(api_key="sk-...")
cached_gpt = CachedLLMClient(gpt_client) # 给任何客户端加缓存
2. 父类过于庞大（God Base Class）
早期我写过一个BaseLLMClient，包含了日志、缓存、重试、监控、序列化… 结果这个类有2000多行。子类继承时痛苦不堪，因为父类一改，所有子类都可能受影响。
经验法则：如果发现父类方法超过15个，或者子类经常要super().xxx()调用父类方法，就该考虑拆分了。用混入类（Mixin） 代替：
class LoggingMixin:
"""只负责日志的混入类"""
def _log_call(self, prompt, result):
self.logger.info(f"调用{self.__class__.__name__}: {prompt[:30]}...")
class RetryMixin:
"""只负责重试的混入类"""
def call_with_retry(self, prompt, max_retries=3):
for i in range(max_retries):
try:
return self.call(prompt)
except Exception as e:
if i == max_retries - 1:
raise
self.logger.warning(f"第{i+1}次重试: {e}")
# 按需组合
class ProductionGPT4Client(LoggingMixin, RetryMixin, GPT4Client):
"""生产环境用的GPT-4客户端，带了日志和重试"""
pass
构建LLM应用框架的实战模式
基于继承和多态，我们可以设计一个可扩展的LLM应用框架：
from abc import ABC, abstractmethod
from typing import List, Dict, Any
class BaseLLMComponent(ABC):
"""框架中所有组件的基类"""
@abstractmethod
def process(self, input_data: Any) -> Any:
pass
def __str__(self):
return f"{self.__class__.__name__}"
class PromptTemplate(BaseLLMComponent):
"""提示词模板组件"""
def __init__(self, template: str):
self.template = template
def process(self, context: Dict) -> str:
# 简单的模板渲染
return self.template.format(**context)
class OutputParser(BaseLLMComponent):
"""输出解析组件"""
def process(self, raw_response: str) -> Dict:
# 这里可以解析JSON、提取关键信息等
return {"raw": raw_response, "parsed": raw_response.strip()}
class LLMOrchestrator:
"""编排器——框架的核心"""
def __init__(self):
self.components: List[BaseLLMComponent] = []
def add_component(self, component: BaseLLMComponent):
"""添加任意组件，只要它继承自BaseLLMComponent"""
self.components.append(component)
def run_pipeline(self, initial_input: Any) -> Any:
"""运行整个处理流水线"""
current_data = initial_input
for component in self.components:
print(f"正在执行: {component}")
current_data = component.process(current_data)
return current_data
# 使用框架构建应用
orchestrator = LLMOrchestrator()
# 添加各种组件（多态的魅力）
orchestrator.add_component(PromptTemplate("请用{style}风格回答：{question}"))
orchestrator.add_component(GPT4Client(api_key="sk-...")) # 这也是BaseLLMComponent
orchestrator.add_component(OutputParser())
# 运行流水线
context = {"style": "幽默", "question": "什么是继承？"}
result = orchestrator.run_pipeline(context)
这个框架的妙处在于：LLMOrchestrator完全不知道它处理的组件具体是什么，只知道它们都有process方法。明天你想加个“敏感词过滤组件”，写个SensitiveFilter继承BaseLLMComponent，然后add_component就行——开闭原则的完美体现：对扩展开放，对修改关闭。
给工程师的几点经验建议
-
继承是“是一个”关系，不是“有一个”关系。
CachedLLMClient“有一个”LLM客户端（组合），而不是“是一个”LLM客户端（继承）。判断错了，设计就歪了。 -
多态不一定要用继承。Python的鸭子类型意味着只要对象有需要的方法，它就能工作。这是Python比Java等语言灵活的地方，别被传统OOP教条束缚。
-
在LLM应用开发中，把变化的部分封装成独立的类。模型会变（GPT-3.5/4/5…），提示词模板会变，输出解析方式会变——每个变化点都是一个潜在的类。
-
调试技巧：当你的继承层次超过3层，就该停下来想想是不是设计复杂了。好的继承结构像树，坏的继承结构像网。
-
实用主义优先：小型脚本直接用if-else没问题；超过500行的项目，才需要考虑这些设计模式。别为了“设计模式”而过度设计。
最后记住，这些技术手段的目标只有一个：让代码更容易应对变化。在LLM这个每天都有新模型、新API的领域，可扩展性不是奢侈品，是生存必需品。
下一篇预告：010、异常处理与调试——让LLM应用在半夜也能稳定运行# 010、综合实战：用扎实的Python基础构建一个简易对话Agent
昨天在调试一个多轮对话逻辑时，又遇到了那个老问题——状态丢失。每次用户换个话题，之前的上下文就像没存在过一样。这让我想起刚入行时写的第一个聊天机器人，用一堆全局变量拼凑状态，结果代码像一团乱麻。今天咱们就用最基础的Python特性，重新构建一个清晰可维护的对话Agent，你会发现，那些看似简单的列表、字典、函数和类，用对了地方就能产生奇效。
从混乱的字典到结构化的Agent
最初版本的对话管理大概是这样的：
# 别这样写！状态散落各处，难以追踪
user_history = []
current_topic = None
pending_questions = []
这种写法在简单场景下还能凑合，一旦需要多个用户同时对话，或者要保存对话状态，立刻就会暴露问题。咱们需要的是一种封装良好的方式。
用类封装对话状态
先定义一个最简单的Agent类，把状态收拢到一起：
class SimpleAgent:
def __init__(self, name="助理"):
self.name = name # Agent名称
self.memory = [] # 对话历史，用列表存储
self.context = {} # 当前上下文，字典最合适
self.tools = {} # 可用工具函数注册在这里
def add_memory(self, role, content):
"""记录对话片段，role区分用户和AI"""
self.memory.append({
"role": role,
"content": content,
"timestamp": len(self.memory) # 简单用序号代替时间戳
})
# 控制记忆长度，避免无限增长
if len(self.memory) > 20:
self.memory = self.memory[-10:] # 只保留最近10条
print(f"[Debug] 记忆已裁剪，现在保留{len(self.memory)}条记录")
这里有个细节：self.memory用列表而不是字典，因为对话记录本质上是顺序结构。而self.context用字典，因为上下文是键值对形式。选择合适的数据结构，代码会自然变得清晰。
工具注册与调用机制
真正的Agent需要能调用外部功能。咱们用字典来管理工具集，函数名作为key，函数本身作为value：
def register_tool(self, func):
"""注册工具函数，装饰器风格"""
self.tools[func.__name__] = func
return func # 返回原函数，保持装饰器特性
def has_tool(self, tool_name):
"""检查工具是否存在，避免运行时崩溃"""
return tool_name in self.tools
def call_tool(self, tool_name, **kwargs):
"""安全调用工具，这里踩过坑——一定要做异常处理"""
if not self.has_tool(tool_name):
return f"错误：找不到工具 {tool_name}"
try:
result = self.tools[tool_name](**kwargs)
# 记录工具调用到上下文
self.context[f"last_tool_{tool_name}"] = result
return result
except Exception as e:
return f"工具调用失败：{str(e)}"
注意**kwargs的使用，它让工具函数可以灵活接收任意参数。这种设计模式在实际开发中很常见，特别是当你需要动态调用不同签名的函数时。
对话处理的核心逻辑
现在来实现最关键的process方法，它展示了如何综合运用基础语法：
def process(self, user_input):
"""处理用户输入，返回Agent响应"""
# 1. 记录用户输入
self.add_memory("user", user_input)
# 2. 更新上下文（提取关键信息）
self._update_context(user_input)
# 3. 判断是否需要调用工具
tool_to_use = self._detect_tool_call(user_input)
if tool_to_use:
# 从输入中提取参数，这里简化处理
params = self._extract_params(user_input)
tool_result = self.call_tool(tool_to_use, **params)
# 将工具结果也加入记忆
self.add_memory("system", f"调用工具 {tool_to_use}，结果：{tool_result}")
# 4. 生成响应（这里用简单规则代替真实LLM）
response = self._generate_response()
# 5. 记录Agent响应
self.add_memory("assistant", response)
return response
def _update_context(self, text):
"""从输入中提取关键信息更新上下文"""
# 实际项目这里会用NLU组件，咱们简单实现
if "天气" in text:
self.context["current_topic"] = "weather"
elif "时间" in text:
self.context["current_topic"] = "time"
# 记录最后对话时间（模拟）
self.context["last_interaction"] = len(self.memory)
def _detect_tool_call(self, text):
"""检测是否需要调用工具"""
# 简单的关键词匹配，实际应该用更智能的方式
tool_keywords = {
"天气": "get_weather",
"时间": "get_time",
"计算": "calculate"
}
for keyword, tool_name in tool_keywords.items():
if keyword in text and self.has_tool(tool_name):
return tool_name
return None
看到没？一个完整的处理流程，用的全是列表追加、字典查找、函数调用这些基础操作。复杂系统往往就是这些简单元素的有机组合。
实际使用示例
让我们实例化一个Agent并添加一些实用工具：
# 创建Agent实例
agent = SimpleAgent(name="小助手")
# 注册工具（用装饰器语法很优雅）
@agent.register_tool
def get_weather(city="北京"):
"""模拟天气查询，实际会调用API"""
weather_data = {
"北京": "晴，25℃",
"上海": "多云，27℃",
"深圳": "阵雨，30℃"
}
return weather_data.get(city, "未知城市")
@agent.register_tool
def calculate(expression):
"""安全计算表达式，用eval要小心！"""
try:
# 生产环境千万别直接eval用户输入！
# 这里仅演示，实际要用ast.literal_eval或解析库
allowed_chars = set("0123456789+-*/. ()")
if all(c in allowed_chars for c in expression):
return str(eval(expression))
return "非法表达式"
except:
return "计算失败"
# 开始对话
print(agent.process("今天北京天气怎么样？"))
print(agent.process("那上海呢？")) # 这里能利用上下文
print(agent.process("计算一下3+5*2"))
print(agent.process("我们刚才聊了什么？")) # 可以查询历史
最后那个查询历史的功能，我们可以在_generate_response里实现：
def _generate_response(self):
"""生成响应，根据上下文和历史决定回答"""
if "current_topic" not in self.context:
return "你好！我可以帮你查询天气、时间或进行计算。"
topic = self.context["current_topic"]
# 根据话题生成不同响应
if topic == "weather":
last_tool = self.context.get("last_tool_get_weather", "未知")
return f"天气信息：{last_tool}"
# 简单实现历史查询
if len(self.memory) > 0 and "聊了什么" in self.memory[-1]["content"]:
# 返回最近3条记录
recent = self.memory[-3:] if len(self.memory) >= 3 else self.memory
summary = "\n".join([f"{m['role']}: {m['content']}" for m in recent])
return f"最近对话：\n{summary}"
return "我已经处理了你的请求。"
一些踩坑经验
调试这个简单Agent时，我重新确认了几个老经验：
第一，状态管理要尽早抽象。不要等到代码里散落着十几个全局变量才想起来封装。看到三个以上相关变量，就该考虑用类组织起来。
第二，字典用.get()方法比直接键访问安全得多。那些KeyError异常很多时候可以通过一个默认值避免，特别是在处理用户输入或外部数据时。
第三，工具函数的注册机制值得投入设计时间。早期我总喜欢用if-elif链判断调用哪个函数，后来发现用字典映射函数名到函数对象，代码扩展性好了不止一个量级。新加功能只需要注册新函数，主流程完全不用动。
第四，对话历史一定要设上限。列表会一直增长，内存泄漏往往从这里开始。实际项目中可能要用更智能的摘要机制，但最简单的长度限制也能解决80%的问题。
最后，这种基础练习的价值在于理解组合的威力。列表、字典、函数、类，每个都不复杂，但像乐高一样组合起来，就能搭建出实用的系统。下次当你面对复杂需求时，不妨先拆解成这些基础元素，往往能找到清晰的实现路径。
这个简易Agent还有很多可扩展方向：添加持久化存储、集成真实LLM、支持多轮任务拆解。但无论怎么扩展，底层还是这些Python基础构件在支撑。扎实的基础不是知道多少高级特性，而是能把简单特性用到恰到好处。
想要解锁更多大模型工程化实战：从部署到落地全流程附源码+避坑指南》，欢迎订阅我的专栏。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
15
0- 0
已为社区贡献2条内容
所有评论(0)