LearnGraph.online

Appearance

1.3 装饰器：LangChain 的核心模式 ​

引言：给函数"穿上盔甲" ​

小白理解：装饰器是 Python 最"神奇"的特性之一，但其实原理很简单！

想象你有一个普通士兵（函数），你想让他变得更强，有两种方法：

  1. 改造士兵本身：给他做手术、换器官（修改原函数代码）
  2. 给士兵穿装备：穿盔甲、拿武器（使用装饰器）

装饰器就是第二种方法——不改变士兵本身，只是给他穿上装备！

如果你阅读过 LangChain 的源码，你会发现装饰器无处不在：

python
@tool
def search_web(query: str) -> str:
    """搜索网络"""
    pass

@chain
def my_chain(input: dict) -> dict:
    """自定义链"""
    pass

@retry(max_attempts=3)
def call_api():
    """调用 API"""
    pass

装饰器是 Python 中最优雅、最强大的特性之一。它能让你在不修改原函数代码的情况下，为函数添加新功能。这就像给函数"穿上盔甲"，让它具备额外的能力。

学习目标 ​

  • ✅ 理解装饰器的本质：函数的包装器
  • ✅ 编写自定义装饰器
  • ✅ 使用 functools.wraps 保留元数据
  • ✅ 实现带参数的装饰器
  • ✅ 理解 LangChain 的 @tool 装饰器
  • ✅ 实战：实现 @retry 和 @log 装饰器

第一部分：装饰器的本质 ​

为什么需要装饰器？ ​

小白解读：假设你有 100 个函数，现在老板说"每个函数执行前都要打印日志"。

笨方法：打开每个函数，在开头加一行 print("开始执行")

  • 要改 100 个地方！
  • 如果以后要改，又要改 100 个地方！

聪明方法：写一个装饰器，往每个函数上一贴就行

python
@log_execution  # 贴上这个标签
def my_function():
    ...

装饰器 = 批量给函数加功能的神器！

装饰器就是高阶函数 ​

python
def my_decorator(func):
    """最简单的装饰器"""
    def wrapper():
        print("函数执行前")
        func()
        print("函数执行后")
    return wrapper

# 手动装饰
def say_hello():
    print("Hello!")

say_hello = my_decorator(say_hello)
say_hello()

# 输出:
# 函数执行前
# Hello!
# 函数执行后

图解装饰器原理：

原始的 say_hello 函数：
┌─────────────────┐
│  print("Hello!")│
└─────────────────┘

经过 my_decorator 包装后：
┌─────────────────────────────┐
│  print("函数执行前")         │ ← 新增的功能
│  ┌─────────────────┐        │
│  │  print("Hello!")│        │ ← 原函数
│  └─────────────────┘        │
│  print("函数执行后")         │ ← 新增的功能
└─────────────────────────────┘

关键：原函数代码一行没改，但现在有了新功能！

使用 @ 语法 ​

python
@my_decorator
def say_hello():
    print("Hello!")

# 等价于: say_hello = my_decorator(say_hello)

say_hello()

💡 关键概念：@decorator 只是语法糖，本质是函数调用。

小白理解 - @ 语法的秘密：

当你写：

python
@my_decorator
def say_hello():
    print("Hello!")

Python 实际上执行的是：

python
def say_hello():
    print("Hello!")
say_hello = my_decorator(say_hello)  # 把函数"喂"给装饰器，拿回增强版

@ 符号只是一个"快捷方式"！ 让代码更简洁。

第二部分：处理参数和返回值 ​

处理任意参数 ​

python
def timer_decorator(func):
    """计时装饰器"""
    import time
    
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} 执行时间: {end - start:.4f}秒")
        return result
    
    return wrapper

@timer_decorator
def calculate_sum(n: int) -> int:
    """计算 1 到 n 的和"""
    total = sum(range(1, n + 1))
    return total

result = calculate_sum(1000000)
print(f"结果: {result}")
# calculate_sum 执行时间: 0.0234秒
# 结果: 500000500000

保留函数元数据 ​

python
from functools import wraps

def my_decorator(func):
    @wraps(func)  # 关键！保留原函数的元数据
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@my_decorator
def example():
    """这是示例函数"""
    pass

print(example.__name__)  # example（而不是 wrapper）
print(example.__doc__)   # 这是示例函数

⚠️ 重要：始终使用 @wraps(func)，否则会丢失函数的名称、文档等信息。

小白理解 - 为什么需要 @wraps？

问题：装饰后的函数"忘记"了自己是谁

python
@my_decorator
def example():
    """这是示例函数"""
    pass

print(example.__name__)  # 输出 "wrapper"！不是 "example"！
print(example.__doc__)   # 输出 None！文档丢了！

解决：用 @wraps(func) 保留身份信息

python
from functools import wraps

def my_decorator(func):
    @wraps(func)  # 告诉 wrapper："你要伪装成 func"
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

这样 example.__name__ 就是 "example"，文档也保留了！

第三部分：实用装饰器实现 ​

1. 日志装饰器 ​

python
from functools import wraps
from typing import Callable
import logging

logging.basicConfig(level=logging.INFO)

def log_execution(func: Callable) -> Callable:
    """
    记录函数执行的装饰器
    
    Args:
        func: 要装饰的函数
    
    Returns:
        包装后的函数
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        logging.info(f"调用函数: {func.__name__}")
        logging.info(f"参数: args={args}, kwargs={kwargs}")
        
        try:
            result = func(*args, **kwargs)
            logging.info(f"返回值: {result}")
            return result
        except Exception as e:
            logging.error(f"异常: {e}")
            raise
    
    return wrapper

@log_execution
def call_llm(prompt: str, model: str = "gpt-3.5-turbo") -> str:
    """调用 LLM API"""
    return f"[模拟响应] {prompt[:20]}..."

result = call_llm("What is AI?", model="gpt-4")

2. 重试装饰器 ​

python
import time
from functools import wraps
from typing import Callable

def retry(
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0
) -> Callable:
    """
    带指数退避的重试装饰器
    
    Args:
        max_attempts: 最大尝试次数
        delay: 初始延迟（秒）
        backoff: 退避因子
    
    Returns:
        装饰器函数
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            current_delay = delay
            
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    
                    print(f"尝试 {attempt + 1} 失败: {e}")
                    print(f"等待 {current_delay:.2f}秒后重试...")
                    time.sleep(current_delay)
                    current_delay *= backoff
        
        return wrapper
    return decorator

@retry(max_attempts=3, delay=1.0, backoff=2.0)
def unstable_api_call():
    """模拟不稳定的 API 调用"""
    import random
    if random.random() < 0.7:
        raise ConnectionError("API 调用失败")
    return "成功"

# 测试
try:
    result = unstable_api_call()
    print(f"最终结果: {result}")
except Exception as e:
    print(f"所有重试均失败: {e}")

3. 缓存装饰器 ​

python
from functools import wraps, lru_cache
from typing import Callable

def cache_result(func: Callable) -> Callable:
    """
    简单的结果缓存装饰器
    
    Args:
        func: 要装饰的函数
    
    Returns:
        带缓存的函数
    """
    cache = {}
    
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 创建缓存键
        key = str(args) + str(kwargs)
        
        if key in cache:
            print(f"从缓存返回: {func.__name__}")
            return cache[key]
        
        result = func(*args, **kwargs)
        cache[key] = result
        return result
    
    return wrapper

@cache_result
def expensive_computation(n: int) -> int:
    """模拟耗时计算"""
    print(f"计算 fibonacci({n})...")
    if n <= 1:
        return n
    return expensive_computation(n-1) + expensive_computation(n-2)

# Python 内置的缓存装饰器（更高效）
@lru_cache(maxsize=128)
def fibonacci(n: int) -> int:
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

第四部分：LangChain 的 @tool 装饰器 ​

理解 @tool 的工作原理 ​

python
# LangChain 中的 Tool 装饰器简化实现
from functools import wraps
from typing import Callable, Any

def tool(name: str = None, description: str = None):
    """
    模拟 LangChain 的 @tool 装饰器
    
    Args:
        name: 工具名称
        description: 工具描述
    
    Returns:
        装饰器函数
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            # 这里可以添加验证、日志等
            result = func(*args, **kwargs)
            return result
        
        # 添加元数据
        wrapper.tool_name = name or func.__name__
        wrapper.tool_description = description or func.__doc__ or ""
        wrapper.is_tool = True
        
        return wrapper
    
    return decorator

# 使用示例
@tool(name="web_search", description="搜索网络并返回结果")
def search_web(query: str) -> str:
    """
    搜索网络
    
    Args:
        query: 搜索查询
    
    Returns:
        搜索结果
    """
    return f"搜索结果: {query}"

# 访问元数据
print(f"工具名: {search_web.tool_name}")
print(f"工具描述: {search_web.tool_description}")
print(f"是否为工具: {search_web.is_tool}")

真实的 LangChain Tool 示例 ​

python
from langchain.tools import tool

@tool
def get_weather(location: str) -> str:
    """
    获取指定地点的天气信息
    
    Args:
        location: 地点名称
    
    Returns:
        天气信息
    """
    # 实际应用中会调用天气 API
    return f"{location} 的天气：晴天，25°C"

@tool
def calculate(expression: str) -> str:
    """
    计算数学表达式
    
    Args:
        expression: 数学表达式
    
    Returns:
        计算结果
    """
    try:
        result = eval(expression)
        return f"结果: {result}"
    except Exception as e:
        return f"计算错误: {e}"

# 在 Agent 中使用
# from langgraph.prebuilt import create_react_agent
# agent = create_react_agent(
#     model=llm,
#     tools=[get_weather, calculate]
# )

第五部分：组合多个装饰器 ​

装饰器堆叠 ​

python
@log_execution
@retry(max_attempts=3)
@cache_result
def critical_api_call(param: str) -> str:
    """关键的 API 调用"""
    return f"Result for {param}"

# 等价于:
# critical_api_call = log_execution(retry(cache_result(critical_api_call)))

# 执行顺序：
# 1. 先应用 cache_result
# 2. 然后应用 retry
# 3. 最后应用 log_execution

⚠️ 注意顺序：装饰器从下往上应用！

第六部分：实战 - 完整的装饰器库 ​

python
"""
Agent 装饰器库
为 AI Agent 函数提供常用装饰器
"""

import time
import functools
import logging
from typing import Callable, Any

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


def measure_time(func: Callable) -> Callable:
    """测量函数执行时间"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> Any:
        start = time.perf_counter()
        result = func(*args, **kwargs)
        end = time.perf_counter()
        logger.info(f"{func.__name__} 执行时间: {end - start:.4f}秒")
        return result
    return wrapper


def log_calls(func: Callable) -> Callable:
    """记录函数调用"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> Any:
        logger.info(f"调用 {func.__name__}")
        logger.debug(f"参数: args={args}, kwargs={kwargs}")
        result = func(*args, **kwargs)
        logger.debug(f"返回: {result}")
        return result
    return wrapper


def validate_args(*validators):
    """
    参数验证装饰器
    
    Args:
        *validators: 验证函数列表
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            for validator in validators:
                if not validator(*args, **kwargs):
                    raise ValueError(f"参数验证失败: {func.__name__}")
            return func(*args, **kwargs)
        return wrapper
    return decorator


def rate_limit(calls: int, period: int):
    """
    速率限制装饰器
    
    Args:
        calls: 允许的调用次数
        period: 时间周期（秒）
    """
    from collections import deque
    
    def decorator(func: Callable) -> Callable:
        timestamps = deque(maxlen=calls)
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            now = time.time()
            
            if len(timestamps) == calls:
                time_since_oldest = now - timestamps[0]
                if time_since_oldest < period:
                    sleep_time = period - time_since_oldest
                    logger.warning(f"速率限制: 等待 {sleep_time:.2f}秒")
                    time.sleep(sleep_time)
            
            timestamps.append(now)
            return func(*args, **kwargs)
        
        return wrapper
    return decorator


# 使用示例
@measure_time
@log_calls
@rate_limit(calls=3, period=10)
def call_llm_api(prompt: str) -> str:
    """调用 LLM API"""
    logger.info(f"处理提示: {prompt}")
    time.sleep(0.5)  # 模拟 API 延迟
    return f"响应: {prompt}"


def main():
    """演示装饰器使用"""
    for i in range(5):
        result = call_llm_api(f"请求 #{i+1}")
        print(result)


if __name__ == "__main__":
    main()

本节总结 ​

核心概念 ​

  1. 装饰器 = 高阶函数 - 接受函数，返回增强的函数
  2. @wraps 必不可少 - 保留原函数元数据
  3. 装饰器可以带参数 - 返回装饰器的函数
  4. 装饰器可以堆叠 - 从下往上应用
  5. LangChain 大量使用装饰器 - @tool, @chain 等

核心概念一览表 ​

概念一句话解释生活比喻
装饰器不修改原函数，添加新功能给士兵穿盔甲
@语法装饰器的简写形式快捷键
wrapper包装函数，执行额外逻辑盔甲本身
@wraps保留原函数身份信息盔甲上写着士兵的名字
带参数装饰器可配置的装饰器可调节的盔甲
堆叠装饰器多个装饰器叠加穿多层装备

新手常见问题 ​

Q1：为什么我的装饰器函数名变成了 wrapper？

  • 忘记加 @wraps(func)，加上就好了

Q2：带参数的装饰器为什么要三层嵌套？

  • 外层接收装饰器参数
  • 中层接收被装饰的函数
  • 内层是实际执行的 wrapper

Q3：装饰器堆叠的顺序重要吗？

  • 重要！从下往上应用。最下面的装饰器最先包装函数

常用装饰器模式 ​

装饰器用途应用场景
@log_execution记录调用调试、审计
@retry自动重试API 调用
@cache_result结果缓存昂贵计算
@measure_time性能分析优化
@rate_limit速率限制API 限流
@toolTool 注册LangChain Agent

下一节：1.4 模块与包管理

基于 MIT 许可证发布。内容版权归作者所有。

Copyright © 2025-present 王几行XING（Bryce Wang）