4. Python基础回顾：面向对象、装饰器、生成器、异步编程等与大模型开发相关的核心语法

大模型应用开发并非从零开始，而是建立在Python语言核心特性之上的工程实践。无论是封装一个LLM客户端、设计一个提示词模板链，还是实现流式输出、并发调用多个模型接口，都离不开面向对象、装饰器、生成器与异步编程这四大语法支柱。本章将聚焦于这些与大模型开发强相关的核心语法，通过原理剖析、代码案例与常见误区复盘，帮助你建立从“会写Python”到“能开发大模型应用”的语法桥梁。

上图展示了四大语法支柱与大模型应用开发之间的映射关系。面向对象提供了结构化封装的基础，装饰器解决了横切关注点，生成器天然适配流式输出，异步编程则让IO密集型的大模型调用变得高效。下面我们逐一深入。

4.1 面向对象：构建大模型应用的骨架

核心原理

面向对象编程（OOP）的核心思想是将数据（属性）与操作数据的方法（行为）绑定为一个整体——对象。在大模型开发中，一个LLM客户端通常包含、base_url、model_name等属性，以及chat()、embed()等方法。通过类（Class）可以创建多个独立实例，避免全局变量污染，同时利用继承和多态实现不同模型提供商的统一接口。

判断方法：何时使用类？

• 当需要维护一组关联状态（如模型配置、历史对话、计数）时。
• 当多个函数需要共享同一份配置或资源时。
• 当需要为不同模型提供商（、Anthropic、本地模型）提供统一接口时。

案例拆解：设计一个可扩展的LLM客户端

class BaseLLMClient:
    """所有模型客户端的基类，定义统一接口"""
    def __init__(self, : str, base_url: str, model: str):
        self. = 
        self.base_url = base_url
        self.model = model
        self.history = []  # 保存对话历史

    def chat(self, messages: list) -> str:
        raise NotImplementedError("子类必须实现chat方法")

    def add_to_history(self, role: str, content: str):
        self.history.append({"role": role, "content": content})

class OpenAIClient(BaseLLMClient):
    """ 兼容接口实现"""
    def chat(self, messages: list) -> str:
        # 实际调用  SDK 或 HTTP API
        # 这里仅作示意
        return f"[ {self.model}] 模拟响应: {messages[-1]['content']}"

class LocalClient(BaseLLMClient):
    """本地部署模型（如 Ollama）实现"""
    def chat(self, messages: list) -> str:
        return f"[Local {self.model}] 模拟响应: {messages[-1]['content']}"

# 使用示例
client = OpenAIClient(="sk-xxx", base_url="https://api..com/v1", model="")
client.add_to_history("user", "你好")
response = client.chat(client.history)
print(response)

这个案例展示了OOP的核心价值：基类定义契约，子类实现细节。当需要接入新的模型提供商时，只需继承BaseLLMClient并实现chat()方法，无需修改调用方代码。同时，history属性封装了对话状态，避免了全局变量。

常见误区与复盘

• 误区1：过度设计类层次。对于简单脚本，直接使用函数和字典可能更清晰。判断标准：如果类只有1个方法且无状态，考虑使用函数。
• 误区2：忽略组合优于继承。当“有一个”关系（如客户端有一个HTTP会话）时，应使用组合而非继承。
• 误区3：在__init__中执行耗时操作（如网络连接）。应延迟初始化或使用懒加载模式。

4.2 装饰器：为模型调用注入横切能力

核心原理

装饰器是一种高阶函数，它接收一个函数作为参数，返回一个增强后的函数。其本质是func = decorator(func)的语法糖。在大模型开发中，装饰器常用于：重试机制（应对API限流）、缓存（避免重复计算）、日志记录（追踪调用）、鉴权校验（保护内部接口）。

操作流程：编写一个带参数的重试装饰器

import time
import functools

def retry(max_retries: int = 3, delay: float = 1.0, exceptions: tuple = (Exception,)):
    """重试装饰器，支持指定异常类型和重试次数"""
    def decorator(func):
        @functools.wraps(func)  # 保留原函数元信息
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_retries - 1:
                        raise  # 最后一次重试失败，抛出原始异常
                    print(f"第{attempt+1}次调用失败: {e}, {delay}秒后重试...")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

# 应用示例
@retry(max_retries=3, delay=2.0, exceptions=(TimeoutError, ConnectionError))
def call_llm_api(prompt: str) -> str:
    # 模拟可能失败的API调用
    import random
    if random.random() < 0.5:
        raise TimeoutError("模拟超时")
    return f"成功响应: {prompt}"

print(call_llm_api("你好"))

案例拆解：缓存装饰器提升性能

在Prompt模板渲染或Embedding计算中，相同输入往往产生相同输出。使用functools.lru_cache可以自动缓存函数结果：

from functools import lru_cache

@lru_cache(maxsize=128)
def get_embedding(text: str) -> list:
    """模拟获取文本向量，实际中会调用Embedding API"""
    print(f"计算embedding: {text[:20]}...")
    return [len(text), sum(ord(c) for c in text) % 100]

# 第一次调用会执行函数体
vec1 = get_embedding("大模型")
# 第二次调用直接命中缓存
vec2 = get_embedding("大模型")
print(vec1 == vec2)  # True

常见误区与复盘

• 误区1：忘记使用functools.wraps，导致被装饰函数的__name__和__doc__丢失，影响调试。
• 误区2：装饰器内层函数未返回原函数的返回值，导致调用结果变为None。
• 误区3：多个装饰器叠加时顺序错误。装饰器从下往上应用，执行时从上往下。例如@auth @retry会先执行auth再执行retry。

4.3 生成器：实现流式输出与内存友好

核心原理

生成器是包含yield关键字的函数，调用时返回一个迭代器对象。与普通函数一次性返回所有结果不同，生成器惰性求值，每次迭代只计算下一个值。在大模型场景中，流式输出（Streaming）是标配功能——模型逐生成内容，客户端逐字显示，极大提升用户体验。

操作流程：将API流式响应封装为生成器

def stream_chat(prompt: str, chunk_size: int = 5):
    """模拟流式返回文本块"""
    full_response = f"这是关于'{prompt}'的完整回答，包含多个句子。"
    for i in range(0, len(full_response), chunk_size):
        yield full_response[i:i+chunk_size]
        # 模拟网络延迟
        import time; time.sleep(0.1)

# 使用生成器逐块处理
for chunk in stream_chat("Python"):
    print(chunk, end="", flush=True)
print("\n---流式输出完成---")

案例拆解：生成器在数据处理管道中的应用

在构建RAG（检索增强生成）应用时，需要处理大量文档。生成器可以构建高效的数据处理管道：

def read_large_file(file_path: str):
    """逐行读取大文件，避免一次性加载到内存"""
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            yield line.strip()

def filter_keywords(lines, keywords: set):
    """过滤包含关键词的行"""
    for line in lines:
        if any(kw in line for kw in keywords):
            yield line

def process_pipeline(file_path: str, keywords: set):
    """组合多个生成器形成管道"""
    lines = read_large_file(file_path)
    filtered = filter_keywords(lines, keywords)
    for line in filtered:
        # 模拟进一步处理（如向量化）
        yield f"处理: {line[:50]}"

# 执行管道
for result in process_pipeline("data.txt", {"大模型", "AI"}):
    print(result)

这个案例展示了生成器的管道模式：每个生成器只负责一个步骤，数据在迭代时按需流动，内存占用恒定。这对于处理超过内存容量的语料库至关重要。

常见误区与复盘

• 误区1：混淆生成器与普通函数。生成器函数体不会立即执行，只有迭代时才执行。
• 误区2：生成器只能迭代一次。如果需要重复使用，需重新调用生成器函数。
• 误区3：在生成器中使用return返回值会被忽略，return仅用于终止迭代。

4.4 异步编程：并发调用大模型API

核心原理

异步编程（async/await）允许在等待IO操作（如网络请求）时让出控制权，执行其他任务。大模型API调用是典型的IO密集型操作，同步调用会阻塞线程，而异步并发可以同时发起多个请求，显著提升吞吐量。Python的asyncio库提供了事件循环和协程支持。

操作流程：并发调用多个模型

import asyncio
import aiohttp

async def call_model(session: aiohttp.ClientSession, model_name: str, prompt: str) -> str:
    """异步调用单个模型"""
    url = f"https://api.example.com/v1/{model_name}/chat"
    payload = {"messages": [{"role": "user", "content": prompt}]}
    async with session.post(url, json=payload) as resp:
        data = await resp.json()
        return data["choices"][0]["message"]["content"]

async def main():
    prompts = ["解释Python", "解释异步", "解释生成器"]
    async with aiohttp.ClientSession() as session:
        # 创建三个并发任务
        tasks = [
            call_model(session, "", prompts[0]),
            call_model(session, "claude-3", prompts[1]),
            call_model(session, "llama-3", prompts[2])
        ]
        # 等待所有任务完成
        results = await asyncio.gather(*tasks)
        for i, res in enumerate(results):
            print(f"模型{i+1}响应: {res[:30]}...")

# 运行事件循环
asyncio.run(main())

案例拆解：带超时与错误处理的并发调用

生产环境中，单个模型可能超时或返回错误。我们需要精细控制并发度和容错：

import asyncio
from asyncio import TimeoutError

async def safe_call(session, model_name, prompt, timeout=5):
    """带超时控制的单次调用"""
    try:
        return await asyncio.wait_for(
            call_model(session, model_name, prompt),
            timeout=timeout
        )
    except TimeoutError:
        return f"[{model_name}] 超时"
    except Exception as e:
        return f"[{model_name}] 错误: {e}"

async def batch_call(prompts: list, concurrency: int = 2):
    """控制并发度的批量调用"""
    semaphore = asyncio.Semaphore(concurrency)
    async def bounded_call(session, prompt):
        async with semaphore:
            return await safe_call(session, "", prompt)
    
    async with aiohttp.ClientSession() as session:
        tasks = [bounded_call(session, p) for p in prompts]
        return await asyncio.gather(*tasks)

# 执行
results = asyncio.run(batch_call(["问题1", "问题2", "问题3", "问题4"], concurrency=2))
print(results)

常见误区与复盘

• 误区1：在异步函数中调用同步阻塞库（如requests），会阻塞整个事件循环。应使用aiohttp或httpx.AsyncClient。
• 误区2：忘记await协程，导致任务未执行。使用asyncio.create_task()创建任务后必须await。
• 误区3：过度使用异步。对于CPU密集型任务（如本地模型推理），异步不会带来性能提升，反而增加复杂度。此时应使用多进程。

4.5 四大语法的协同应用与总结

在实际的大模型应用中，这四种语法往往协同工作。下面是一个综合示例，展示如何构建一个健壮的异步流式客户端：

import asyncio
import functools
from typing import AsyncGenerator

class AsyncLLMClient:
    """异步流式LLM客户端"""
    def __init__(self, : str, model: str):
        self. = 
        self.model = model
        self._session = None

    async def __aenter__(self):
        import aiohttp
        self._session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *args):
        await self._session.close()

    @functools.lru_cache(maxsize=32)
    def _get_cache_key(self, prompt: str) -> str:
        return prompt

    async def stream_chat(self, prompt: str) -> AsyncGenerator[str, None]:
        """异步生成器，流式返回响应"""
        # 模拟异步流式响应
        full_text = f"异步流式响应: {prompt}"
        for i in range(0, len(full_text), 3):
            await asyncio.sleep(0.05)  # 模拟网络延迟
            yield full_text[i:i+3]

async def main():
    async with AsyncLLMClient(="sk-xxx", model="") as client:
        async for chunk in client.stream_chat("你好"):
            print(chunk, end="", flush=True)
        print("\n完成")

asyncio.run(main())

这个综合案例融合了：面向对象（封装客户端）、异步上下文管理器（资源管理）、异步生成器（流式输出）、装饰器（缓存）。这正是大模型应用开发中常见的工程模式。